github-actions[bot] commented on code in PR #65658: URL: https://github.com/apache/doris/pull/65658#discussion_r3796164613
########## be/src/io/cache/async_cache_write_manager.cpp: ########## @@ -0,0 +1,780 @@ +// 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_manager.h" + +#include <algorithm> +#include <array> +#include <exception> +#include <limits> +#include <optional> +#include <thread> +#include <type_traits> +#include <unordered_map> +#include <utility> + +#include "common/exception.h" +#include "common/logging.h" +#include "core/allocator.h" +#include "cpp/sync_point.h" +#include "io/cache/async_cache_write_manager_metrics.h" +#include "io/cache/block_file_cache.h" +#include "runtime/thread_context.h" +#include "util/countdown_latch.h" +#include "util/defer_op.h" +#include "util/time.h" + +namespace doris::io { + +using AsyncCacheWriteAllocator = Allocator<false, false, false, DefaultMemoryAllocator, true>; + +namespace { + +static_assert(std::is_nothrow_move_constructible_v<AsyncCacheWriteTask>); +static_assert(std::is_nothrow_move_assignable_v<AsyncCacheWriteTask>); + +/// Keep an in-progress phase gauge balanced across every return path. +class ScopedActiveCounter { +public: + explicit ScopedActiveCounter(std::atomic<size_t>& counter) : _counter(counter) { + _counter.fetch_add(1, std::memory_order_relaxed); + } + + ~ScopedActiveCounter() { _counter.fetch_sub(1, std::memory_order_relaxed); } + +private: + std::atomic<size_t>& _counter; +}; + +/// Acquire the FIFO mutex while measuring only the actual lock wait and critical-section hold. +class TimedQueueLock { +public: + TimedQueueLock(std::mutex& mutex, bvar::LatencyRecorder& wait_latency, + bvar::LatencyRecorder& hold_latency) + : _lock(mutex, std::defer_lock), + _wait_latency(wait_latency), + _hold_latency(hold_latency) { + const int64_t wait_start_us = MonotonicMicros(); + _lock.lock(); + _acquired_at_us = MonotonicMicros(); + _wait_us = _acquired_at_us - wait_start_us; + } + + ~TimedQueueLock() { + const int64_t hold_us = MonotonicMicros() - _acquired_at_us; + _lock.unlock(); + _wait_latency << _wait_us; + _hold_latency << hold_us; + } + +private: + std::unique_lock<std::mutex> _lock; + bvar::LatencyRecorder& _wait_latency; + bvar::LatencyRecorder& _hold_latency; + int64_t _acquired_at_us {0}; + int64_t _wait_us {0}; +}; + +} // namespace + +CacheAdmissionContext CacheAdmissionContext::from_cache_context(const CacheContext& context, + int64_t tablet_id) { + return CacheAdmissionContext { + .query_id = context.query_id, + .cache_type = context.cache_type, + .expiration_time = context.expiration_time, + .tablet_id = tablet_id, + .is_warmup = context.is_warmup, + }; +} + +CacheContext CacheAdmissionContext::to_cache_context(ReadStatistics* stats) const { + DORIS_CHECK(stats != nullptr); + CacheContext context; + context.query_id = query_id; + context.cache_type = cache_type; + context.expiration_time = expiration_time; + context.tablet_id = tablet_id; + context.is_warmup = is_warmup; + context.stats = stats; + return context; +} + +void AsyncCacheWriteTask::validate() const { + DORIS_CHECK(buffer != nullptr); + DORIS_CHECK(write_epoch.key_token != nullptr); + DORIS_CHECK(write_size > 0); + DORIS_CHECK(write_size <= buffer_size()); + DORIS_CHECK(write_size <= std::numeric_limits<size_t>::max() - file_offset); +} + +size_t AsyncCacheWriteTask::buffer_size() const { + return buffer->size(); +} + +void AsyncCacheWriteTask::finalize() const { + if (on_finalized) { + on_finalized(*this); + } +} + +class AsyncCacheWriteEpochRegistry + : public std::enable_shared_from_this<AsyncCacheWriteEpochRegistry> { +public: + std::shared_ptr<AsyncCacheWriteEpochToken> capture(const UInt128Wrapper& cache_hash) { + auto& shard = _shards[_shard_index(cache_hash)]; + std::lock_guard lock(shard.mutex); + auto iterator = shard.tokens.find(cache_hash); + if (iterator != shard.tokens.end()) { + auto token = iterator->second.token.lock(); + if (token != nullptr) { + return token; + } + shard.tokens.erase(iterator); + _active_key_count.fetch_sub(1, std::memory_order_relaxed); + } + + const uint64_t generation = _next_generation.fetch_add(1, std::memory_order_relaxed); + auto token = std::shared_ptr<AsyncCacheWriteEpochToken>( + new AsyncCacheWriteEpochToken(cache_hash, generation, weak_from_this())); + shard.tokens.emplace(cache_hash, Entry {.generation = generation, .token = token}); + _active_key_count.fetch_add(1, std::memory_order_relaxed); + return token; + } + + void invalidate(const UInt128Wrapper& cache_hash) { + // The token destructor calls release(), so its last strong reference must outlive the shard + // lock instead of re-entering the same mutex from inside this critical section. + std::shared_ptr<AsyncCacheWriteEpochToken> token; + { + auto& shard = _shards[_shard_index(cache_hash)]; + std::lock_guard lock(shard.mutex); + auto iterator = shard.tokens.find(cache_hash); + if (iterator == shard.tokens.end()) { + return; + } + token = iterator->second.token.lock(); + if (token != nullptr) { + token->_valid.store(false, std::memory_order_release); + } + shard.tokens.erase(iterator); + _active_key_count.fetch_sub(1, std::memory_order_relaxed); + } + } + + void release(const UInt128Wrapper& cache_hash, uint64_t generation) { + auto& shard = _shards[_shard_index(cache_hash)]; + std::lock_guard lock(shard.mutex); + auto iterator = shard.tokens.find(cache_hash); + if (iterator == shard.tokens.end() || iterator->second.generation != generation) { + return; + } + shard.tokens.erase(iterator); + _active_key_count.fetch_sub(1, std::memory_order_relaxed); + } + + size_t active_key_count() const { return _active_key_count.load(std::memory_order_relaxed); } + +private: + struct Entry { + uint64_t generation {0}; + std::weak_ptr<AsyncCacheWriteEpochToken> token; + }; + + struct Shard { + std::mutex mutex; + std::unordered_map<UInt128Wrapper, Entry, KeyHash> tokens; + }; + + static constexpr size_t kShardCount = 64; + + static size_t _shard_index(const UInt128Wrapper& cache_hash) { + return KeyHash()(cache_hash) % kShardCount; + } + + std::array<Shard, kShardCount> _shards; + std::atomic<uint64_t> _next_generation {1}; + std::atomic<size_t> _active_key_count {0}; +}; + +AsyncCacheWriteEpochToken::AsyncCacheWriteEpochToken( + const UInt128Wrapper& cache_hash, uint64_t generation, + std::weak_ptr<AsyncCacheWriteEpochRegistry> registry) + : _cache_hash(cache_hash), _generation(generation), _registry(std::move(registry)) {} + +AsyncCacheWriteEpochToken::~AsyncCacheWriteEpochToken() { + auto registry = _registry.lock(); + if (registry != nullptr) { + registry->release(_cache_hash, _generation); + } +} + +Status resolve_async_file_cache_write_max_pending_bytes(int64_t configured_bytes, + int64_t be_mem_limit, + size_t* resolved_bytes) { + DORIS_CHECK(resolved_bytes != nullptr); + if (configured_bytes > 0) { + *resolved_bytes = static_cast<size_t>(configured_bytes); + return Status::OK(); + } + if (configured_bytes != -1) { + return Status::InvalidArgument( + "async file cache write pending byte limit must be positive or -1"); + } + + DORIS_CHECK(be_mem_limit > 0); + constexpr int64_t kMinimumAutoPendingBytes = 1024LL * 1024 * 1024; + *resolved_bytes = static_cast<size_t>(std::max(kMinimumAutoPendingBytes, be_mem_limit / 100)); + return Status::OK(); +} + +class AsyncCacheWriteManager::Worker : public std::enable_shared_from_this<Worker> { +public: + explicit Worker(AsyncCacheWriteManager& manager) : _manager(manager) {} + + Status start() { + auto self = shared_from_this(); + return _manager._worker_pool->submit_func([self = std::move(self)]() { self->_run(); }); + } + + // The caller must hold the manager queue mutex so changing the wait predicate cannot race + // with a worker between evaluating it and blocking on the condition variable. + void request_stop() { _stop_requested.store(true, std::memory_order_release); } + + void wait_until_stopped() { _stopped.wait(); } + +private: + void _run() { + _manager._running_worker_count.fetch_add(1, std::memory_order_relaxed); + Defer mark_finished {[this]() { + const size_t old_running = + _manager._running_worker_count.fetch_sub(1, std::memory_order_relaxed); + DCHECK_GT(old_running, 0); + _stopped.count_down(); + }}; + + while (!_stop_requested.load(std::memory_order_acquire)) { + AsyncCacheWriteTask task; + if (_manager._try_activate_task(&task)) { + _process_task(std::move(task)); + continue; + } + + std::unique_lock lock(_manager._queue_mutex); + TEST_SYNC_POINT("AsyncCacheWriteManager::Worker::_run:before_wait"); + _manager._queue_cv.wait(lock, [this]() { + return !_manager._queue.empty() || _stop_requested.load(std::memory_order_acquire); + }); + } + } + + // A task is the worker loop's exception boundary. Its manager-side completion guard releases + // active accounting and owner state while this boundary keeps the long-lived worker alive. + void _process_task(AsyncCacheWriteTask task) { + const UInt128Wrapper cache_hash = task.cache_hash; + const size_t file_offset = task.file_offset; + const size_t write_size = task.write_size; + try { + _manager._process_task(std::move(task)); + } catch (const Exception& exception) { + _record_task_exception(cache_hash, file_offset, write_size, exception.what()); + } catch (const std::exception& exception) { + _record_task_exception(cache_hash, file_offset, write_size, exception.what()); + } catch (...) { + _record_task_exception(cache_hash, file_offset, write_size, "unknown exception"); + } + } + + void _record_task_exception(const UInt128Wrapper& cache_hash, size_t file_offset, + size_t write_size, const char* message) { + LOG(WARNING) << "Async file cache write task threw an exception, cache=" + << _manager._cache->get_base_path() << ", hash=" << cache_hash.to_string() + << ", offset=" << file_offset << ", size=" << write_size + << ", exception=" << message; + } + + AsyncCacheWriteManager& _manager; + std::atomic<bool> _stop_requested {false}; + CountDownLatch _stopped {1}; +}; + +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); +} + +AsyncCacheWriteManager::AsyncCacheWriteManager(BlockFileCache* cache, + AsyncCacheWriteManagerOptions options) + : _cache(cache), + _options(std::make_shared<const AsyncCacheWriteManagerOptions>(options)), + _write_epoch_registry(std::make_shared<AsyncCacheWriteEpochRegistry>()), + _configured_worker_count(options.worker_count) { + DORIS_CHECK(_cache != nullptr); + DORIS_CHECK(options.worker_count > 0); + DORIS_CHECK(options.max_pending_bytes > 0); + + _mem_tracker = MemTrackerLimiter::create_shared( + MemTrackerLimiter::Type::CACHE, + fmt::format("AsyncFileCacheWrite:{}", _cache->get_base_path())); + _metrics = std::make_unique<Metrics>(*this, _cache->get_base_path().c_str()); +} + +AsyncCacheWriteManager::~AsyncCacheWriteManager() { + shutdown(); +} + +AsyncCacheWriteEpoch AsyncCacheWriteManager::current_write_epoch(const UInt128Wrapper& cache_hash) { + return AsyncCacheWriteEpoch { + .cache_epoch = current_cache_epoch(), + .key_token = _write_epoch_registry->capture(cache_hash), + }; +} + +bool AsyncCacheWriteManager::is_current_write_epoch(const AsyncCacheWriteEpoch& epoch) const { + DORIS_CHECK(epoch.key_token != nullptr); + return epoch.cache_epoch == current_cache_epoch() && epoch.key_token->is_valid(); +} + +bool AsyncCacheWriteManager::check_write_epoch(const AsyncCacheWriteEpoch& epoch) { + DORIS_CHECK(epoch.key_token != nullptr); + if (epoch.cache_epoch != current_cache_epoch()) { + _metrics->record_stale_epoch(Metrics::StaleEpochReason::CACHE); + return false; + } + if (!epoch.key_token->is_valid()) { + _metrics->record_stale_epoch(Metrics::StaleEpochReason::KEY); + return false; + } + return true; +} + +void AsyncCacheWriteManager::invalidate_pending_writes(const UInt128Wrapper& cache_hash) { + _metrics->record_epoch_invalidation(Metrics::EpochInvalidationScope::KEY); + _write_epoch_registry->invalidate(cache_hash); +} + +uint64_t AsyncCacheWriteManager::invalidate_all_pending_writes() { + _metrics->record_epoch_invalidation(Metrics::EpochInvalidationScope::CACHE); + return _cache_epoch.fetch_add(1, std::memory_order_acq_rel) + 1; +} + +size_t AsyncCacheWriteManager::active_write_epoch_key_count() const { + return _write_epoch_registry->active_key_count(); +} + +Status AsyncCacheWriteManager::start() { + std::lock_guard lifecycle_lock(_lifecycle_mutex); + if (!_accepting.load(std::memory_order_acquire)) { + return Status::InternalError("async file cache write manager is shutting down"); + } + if (_started.load(std::memory_order_acquire)) { + return Status::OK(); + } + + const size_t worker_count = _configured_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)); + } + // A failed earlier start may have left a partial worker set. Reconcile the owned workers with + // the latest configured count before publishing readiness. + RETURN_IF_ERROR(_resize_workers_locked(worker_count)); + // 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 AsyncCacheWriteManager::try_submit(AsyncCacheWriteTask task) { + task.validate(); + const int64_t submit_start_us = MonotonicMicros(); + Defer record_submit_latency { + [&]() { _metrics->record_submit_latency(MonotonicMicros() - submit_start_us); }}; + _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("AsyncCacheWriteManager::try_submit:after_register", &task); + if (!_started.load(std::memory_order_acquire) || !_accepting.load(std::memory_order_acquire)) { + _metrics->record_task_rejected(Metrics::RejectionReason::NOT_RUNNING); + return false; + } + + const size_t task_buffer_bytes = task.buffer_size(); + std::optional<AsyncCacheWriteTask> victim; + { + TimedQueueLock lock(_queue_mutex, _metrics->queue_lock_wait_latency(), + _metrics->queue_lock_hold_latency()); + const auto options = _options.load(std::memory_order_acquire); + const size_t max_pending_bytes = options->max_pending_bytes; + const size_t pending_bytes = _pending_bytes.load(std::memory_order_relaxed); + if (_task_buffer_size == 0) { + _task_buffer_size = task_buffer_bytes; + } + DORIS_CHECK(task_buffer_bytes == _task_buffer_size); + + if (task_buffer_bytes > max_pending_bytes) { + _metrics->record_task_rejected(Metrics::RejectionReason::BACKPRESSURE); + return false; + } + + const bool has_capacity = pending_bytes <= max_pending_bytes - task_buffer_bytes; + if (!has_capacity && _queue.empty()) { + _metrics->record_task_rejected(Metrics::RejectionReason::BACKPRESSURE); + return false; + } + + _queue.push_back(std::move(task)); + if (has_capacity) { + _queued_bytes.fetch_add(task_buffer_bytes, std::memory_order_relaxed); + _pending_count.fetch_add(1, std::memory_order_relaxed); + _pending_bytes.fetch_add(task_buffer_bytes, std::memory_order_relaxed); + } else { + victim.emplace(std::move(_queue.front())); + _queue.pop_front(); + } + } + + _metrics->record_task_submitted(task_buffer_bytes); + _queue_cv.notify_one(); + if (victim) { + _complete_task(std::move(*victim), TaskFinalizationReason::EVICTED_OLDEST); + } + return true; +} + +Status AsyncCacheWriteManager::allocate_tracked_buffer(size_t size, + AsyncCacheWriteBufferPtr* buffer) { + DORIS_CHECK(buffer != nullptr); + DORIS_CHECK(size > 0); + const int64_t allocation_start_us = MonotonicMicros(); + Defer record_allocation_latency {[&]() { + _metrics->record_buffer_allocation_latency(MonotonicMicros() - allocation_start_us); + }}; + Status injected_status; + TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteManager::allocate_tracked_buffer:inject_failure", + &injected_status); + if (!injected_status.ok()) { + _metrics->record_buffer_allocation_failure(); + return injected_status; + } + SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_mem_tracker); + Status status = Status::OK(); + try { + ASSIGN_STATUS_IF_CATCH_EXCEPTION(*buffer = AsyncCacheWriteBufferPtr( + new AsyncCacheWriteBuffer(size, _mem_tracker)), + status); + } catch (const std::exception& e) { + status = Status::MemoryAllocFailed("allocate async file cache write buffer failed: {}", + e.what()); + } + if (!status.ok()) { + _metrics->record_buffer_allocation_failure(); + } + return status; +} + +void AsyncCacheWriteManager::_process_task(AsyncCacheWriteTask task) { + Defer complete {[&]() { _complete_active_task(std::move(task)); }}; + + const int64_t age_us = MonotonicMicros() - task.submit_ts_us; + _metrics->record_queue_wait_latency(age_us); + if (!check_write_epoch(task.write_epoch)) { + return; + } + + const int64_t start_us = MonotonicMicros(); + Status status = _persist_task(task); + _metrics->record_worker_task_latency(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.write_size << ", status=" << status; + } +} + +bool AsyncCacheWriteManager::_try_activate_task(AsyncCacheWriteTask* task) { + TimedQueueLock lock(_queue_mutex, _metrics->queue_lock_wait_latency(), + _metrics->queue_lock_hold_latency()); + if (_queue.empty()) { + return false; + } + + *task = std::move(_queue.front()); + _queue.pop_front(); + const size_t task_buffer_bytes = task->buffer_size(); + _queued_bytes.fetch_sub(task_buffer_bytes, std::memory_order_relaxed); + _active_task_count.fetch_add(1, std::memory_order_relaxed); + _active_bytes.fetch_add(task_buffer_bytes, std::memory_order_relaxed); + return true; +} + +Status AsyncCacheWriteManager::_persist_task(const AsyncCacheWriteTask& task) { + if (!check_write_epoch(task.write_epoch)) { + return Status::OK(); + } + + ReadStatistics dummy_stats; + CacheContext context = task.admission_ctx.to_cache_context(&dummy_stats); + auto holder = [&]() { + ScopedActiveCounter active_get_or_set(_active_get_or_set_count); + const int64_t start_us = MonotonicMicros(); + Defer record_latency { + [&]() { _metrics->record_get_or_set_latency(MonotonicMicros() - start_us); }}; + TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteManager::_persist_task:before_get_or_set", &task); + auto result = + _cache->get_or_set(task.cache_hash, task.file_offset, task.write_size, context); + TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteManager::_persist_task:after_get_or_set", &task); + return result; + }(); + + if (!check_write_epoch(task.write_epoch)) { + return Status::OK(); + } + + const size_t task_end = task.file_offset + task.write_size; + for (const auto& block : holder.file_blocks) { + if (block->range().left < task.file_offset || block->range().right >= task_end) { + _metrics->record_skipped_block(Metrics::SkippedBlockReason::PARTIAL_OVERLAP); + continue; + } + if (!check_write_epoch(task.write_epoch)) { + return Status::OK(); + } + if (_cache->is_block_deleting(block)) { + _metrics->record_skipped_block(Metrics::SkippedBlockReason::DELETING); + continue; + } + + switch (block->state()) { + case FileBlock::State::DOWNLOADED: + _metrics->record_skipped_block(Metrics::SkippedBlockReason::DOWNLOADED); + continue; + case FileBlock::State::DOWNLOADING: + _metrics->record_skipped_block(Metrics::SkippedBlockReason::DOWNLOADING); + continue; + case FileBlock::State::SKIP_CACHE: + continue; + case FileBlock::State::EMPTY: + break; + } + + if (block->get_or_set_downloader() != FileBlock::get_caller_id()) { + _metrics->record_skipped_block(Metrics::SkippedBlockReason::DOWNLOADING); + continue; + } + const size_t buffer_offset = block->range().left - task.file_offset; + DORIS_CHECK(buffer_offset <= task.write_size); + DORIS_CHECK(block->range().size() <= task.write_size - buffer_offset); + Status status; + { + ScopedActiveCounter active_append(_active_append_count); + TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteManager::_persist_task:before_append", &task); + const int64_t start_us = MonotonicMicros(); + status = block->append( + Slice(task.buffer->data() + buffer_offset, block->range().size())); + _metrics->record_block_operation_latency(Metrics::BlockOperation::APPEND, + MonotonicMicros() - start_us); + } + if (!status.ok()) { + _metrics->record_block_operation_failure(Metrics::BlockOperation::APPEND); + 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; + } + { + ScopedActiveCounter active_finalize(_active_finalize_count); + const int64_t start_us = MonotonicMicros(); + status = block->finalize(); + _metrics->record_block_operation_latency(Metrics::BlockOperation::FINALIZE, + MonotonicMicros() - start_us); + } + if (!status.ok()) { + _metrics->record_block_operation_failure(Metrics::BlockOperation::FINALIZE); + 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; + continue; + } + _metrics->record_persisted_block(block->range().size()); + } + return Status::OK(); +} + +void AsyncCacheWriteManager::_complete_active_task(AsyncCacheWriteTask task) { + const size_t task_buffer_bytes = task.buffer_size(); + bool became_empty = false; + { + TimedQueueLock lock(_queue_mutex, _metrics->queue_lock_wait_latency(), + _metrics->queue_lock_hold_latency()); + const size_t old_active = _active_task_count.fetch_sub(1, std::memory_order_relaxed); + DCHECK_GT(old_active, 0); + _active_bytes.fetch_sub(task_buffer_bytes, std::memory_order_relaxed); + const size_t old_pending = _pending_count.fetch_sub(1, std::memory_order_relaxed); + DCHECK_GT(old_pending, 0); + _pending_bytes.fetch_sub(task_buffer_bytes, std::memory_order_relaxed); + became_empty = old_pending == 1; + } + _complete_task(std::move(task), TaskFinalizationReason::WORKER_FINISHED); + if (became_empty) { + _queue_cv.notify_all(); + } +} + +void AsyncCacheWriteManager::_complete_task(AsyncCacheWriteTask task, + TaskFinalizationReason reason) { + _metrics->record_task_finalized(task, reason); + task.finalize(); +} + +Status AsyncCacheWriteManager::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 lifecycle_lock(_lifecycle_mutex); + if (!_accepting.load(std::memory_order_acquire)) { + return Status::InternalError("async file cache write manager is shutting down"); + } + if (!_started.load(std::memory_order_acquire)) { + _configured_worker_count.store(worker_count, std::memory_order_release); + return Status::OK(); + } + _configured_worker_count.store(worker_count, std::memory_order_release); + return _resize_workers_locked(worker_count); +} + +Status AsyncCacheWriteManager::_resize_workers_locked(size_t worker_count) { + DORIS_CHECK(_worker_pool != nullptr); + if (worker_count < _workers.size()) { + _stop_workers_locked(worker_count); + RETURN_IF_ERROR(_worker_pool->set_min_threads(static_cast<int>(worker_count))); + RETURN_IF_ERROR(_worker_pool->set_max_threads(static_cast<int>(worker_count))); + return Status::OK(); + } + + RETURN_IF_ERROR(_worker_pool->set_max_threads(static_cast<int>(worker_count))); + // Worker tasks live until resize or shutdown, so reserve one actual pool thread for each task + // before submitting any new Worker. Unlike submit_func(), set_min_threads() propagates an OS + // thread creation failure even when another pool thread is already running; no accepted Worker + // task can therefore remain queued forever without a backing thread. + RETURN_IF_ERROR(_worker_pool->set_min_threads(static_cast<int>(worker_count))); + while (_workers.size() < worker_count) { + auto worker = std::make_shared<Worker>(*this); + RETURN_IF_ERROR(worker->start()); Review Comment: [P1] Make persistent-worker submission transactional across both pool and manager ownership. This call can retain the self-owning runnable before the manager has its only stop handle in two ways: `ThreadPool::do_submit()` first appends it to the token entries and can then throw while publishing the dispatch token, or `start()` can return and the following `_workers.emplace_back()` can throw. In the first case a later option-refresh retry can dispatch the retained unrecorded worker while stranding a newly recorded one; in the second the unrecorded worker may already be running. Shrink/shutdown can stop only `_workers` and can then wait forever. Roll back partial pool submission and establish no-throw manager ownership (or an equivalent cancellable handle) before dispatch; inject failures at both bookkeeping boundaries followed by retry and shutdown. ########## be/src/io/cache/cached_remote_file_reader_async_write.cpp: ########## @@ -0,0 +1,690 @@ +// 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_manager.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/defer_op.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_count"); +bvar::Adder<uint64_t> g_cached_remote_reader_probe_downloaded( + "cached_remote_file_reader_probe_hit_downloaded_count"); +bvar::Adder<uint64_t> g_cached_remote_reader_probe_downloading( + "cached_remote_file_reader_probe_hit_downloading_count"); +bvar::Adder<uint64_t> g_cached_remote_reader_probe_miss( + "cached_remote_file_reader_probe_miss_count"); +bvar::Adder<uint64_t> g_cached_remote_reader_inflight_hit( + "cached_remote_file_reader_inflight_write_buffer_hit_count"); +bvar::Adder<uint64_t> g_cached_remote_reader_async_skip_existing( + "cached_remote_file_reader_async_write_skip_inflight_existing_count"); +bvar::Adder<uint64_t> g_cached_remote_reader_block_wait( + "cached_remote_file_reader_block_wait_count"); +bvar::Adder<uint64_t> g_cached_remote_reader_block_wait_timeout( + "cached_remote_file_reader_block_wait_timeout_count"); +bvar::Adder<uint64_t> g_cached_remote_reader_remote_after_dedup_miss( + "cached_remote_file_reader_remote_read_after_all_dedup_miss_count"); +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"); +bvar::LatencyRecorder g_cached_remote_reader_async_read_plan_latency( + "cached_remote_file_reader_async_read_plan_latency_us"); +bvar::LatencyRecorder g_cached_remote_reader_async_write_submission_latency( + "cached_remote_file_reader_async_write_submission_latency_us"); + +} // namespace + +// One logical cache-block slot in the read plan. +// +// Contract after plan construction: +// - `range.left` is cache-block aligned and ranges are ordered, gap-free, and non-overlapping. +// - `range.size()` is one cache block except for the physical EOF slot, which may be shorter. +// - `source` identifies how this slot can be materialized before the remote read. REMOTE means the +// probe found a real miss; CACHE and DOWNLOADING retain the corresponding probe result by index. +// - `submit_write` is true exactly for REMOTE misses. A CACHE hit lying between the first and last +// miss is reread by the single remote operation, but is deliberately not submitted for writing. +// - `inflight_entry` is non-null exactly when `source` is INFLIGHT and covers the whole slot. +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; +}; + +// Immutable logical partition plus the source classification used by one async read. +// +// `blocks` is an index-stable partition of the aligned read interval: the same index is used for +// inflight lookup, BlockFileCache::probe(), side materialization, remote-buffer slicing, and async +// task creation. The user interval [user_left, user_right] is fully contained by that partition. +// +// `first_remote_block` and `remote_block_count` describe the minimal first-miss-to-last-miss span. +// It may contain CACHE/INFLIGHT hits between misses so the read path issues at most one remote IO. +// `probe_result` is absent only when inflight buffers covered every block and probe was skipped. +// `write_epoch` fences only later persistence: a cache hash identifies immutable file content, so +// an inflight payload remains readable after its original write task is invalidated. +struct CachedRemoteFileReader::AsyncReadPlan { + AsyncReadPlan(AsyncCacheWriteEpoch write_epoch_, size_t user_left_, size_t user_right_) + : write_epoch(std::move(write_epoch_)), + user_left(user_left_), + user_right(user_right_) {} + + AsyncCacheWriteEpoch write_epoch; + 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}; + + // These are construction-time contracts rather than runtime error handling. Keep them as + // debug checks so validating every logical block does not add work to each production read. + void check_layout(size_t file_size, size_t cache_block_size) const { + DCHECK_GT(file_size, 0); + DCHECK_GT(cache_block_size, 0); + DCHECK_LE(user_left, user_right); + DCHECK_LT(user_right, file_size); + DCHECK(!blocks.empty()); + DCHECK_LE(blocks.front().range.left, user_left); + DCHECK_GE(blocks.back().range.right, user_right); + + for (size_t index = 0; index < blocks.size(); ++index) { + const auto& range = blocks[index].range; + DCHECK_LE(range.left, range.right); + DCHECK_EQ(range.left % cache_block_size, 0); + DCHECK_LT(range.right, file_size); + if (index > 0) { + DCHECK_EQ(blocks[index - 1].range.right + 1, range.left); + } + if (range.right + 1 < file_size) { + DCHECK_EQ(range.size(), cache_block_size); + } else { + DCHECK_LE(range.size(), cache_block_size); + } + } + } + + void check_built_sources() const { + size_t first_miss = blocks.size(); + size_t last_miss = 0; + for (size_t index = 0; index < blocks.size(); ++index) { + const auto& block = blocks[index]; + const bool is_inflight = block.source == AsyncReadBlock::Source::INFLIGHT; + const bool is_remote = block.source == AsyncReadBlock::Source::REMOTE; + DCHECK_EQ(block.inflight_entry != nullptr, is_inflight); + DCHECK_EQ(block.submit_write, is_remote); + if (block.submit_write) { + if (first_miss == blocks.size()) { + first_miss = index; + } + last_miss = index; + } + } + + if (first_miss == blocks.size()) { + DCHECK_EQ(remote_block_count, 0); + } else { + DCHECK_EQ(first_remote_block, first_miss); + DCHECK_EQ(remote_block_count, last_miss - first_miss + 1); + } + if (probe_result.has_value()) { + DCHECK_EQ(probe_result->file_blocks.size(), blocks.size()); + } else { + DCHECK(std::all_of(blocks.begin(), blocks.end(), [](const auto& block) { + return block.source == AsyncReadBlock::Source::INFLIGHT; + })); + } + } +}; + +// 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, AsyncCacheWriteEpoch write_epoch, + const IOContext* io_ctx, ReadStatistics& stats) { + const int64_t plan_start_us = MonotonicMicros(); + Defer record_plan_latency {[&]() { + g_cached_remote_reader_async_read_plan_latency << (MonotonicMicros() - plan_start_us); + }}; + DORIS_CHECK(io_ctx != nullptr); + DORIS_CHECK(remaining_offset < size()); + DORIS_CHECK(remaining_size > 0); + DORIS_CHECK(remaining_size <= size() - remaining_offset); + 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); + DORIS_CHECK(align_left % cache_block_size == 0); + DORIS_CHECK(align_left <= remaining_offset); + DORIS_CHECK(align_size > 0); + DORIS_CHECK(align_size <= size() - align_left); + + AsyncReadPlan plan(std::move(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;) { + const size_t block_size = std::min(cache_block_size, align_end - 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); + block_offset += block_size; + } + DORIS_CHECK(!plan.blocks.empty()); + DORIS_CHECK(plan.blocks.front().range.left == align_left); + DORIS_CHECK(plan.blocks.back().range.right == align_end - 1); + plan.check_layout(size(), cache_block_size); + + 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); + 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) { + plan.check_built_sources(); + 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()); + + // probe() validates slot coverage while holding the cache mutex. Use the immutable logical + // plan ranges after it returns: a concurrent file writer may shrink a preallocated EOF block + // during finalize(). + 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) { + 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; + } + plan.check_built_sources(); + 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(user_offset <= plan.user_left); + DORIS_CHECK(plan.user_right - user_offset < result.size); + 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; + const size_t result_offset = copy_left - user_offset; + DORIS_CHECK(copy_size <= result.size - result_offset); + + 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 + result_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); + 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); + TEST_SYNC_POINT("CachedRemoteFileReader::_materialize_async_block:before_wait"); + { + 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 + result_offset, copy_size), + copy_left - read_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=" << read_block.range.left << ", 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); + DORIS_CHECK(plan.first_remote_block <= plan.blocks.size()); + DORIS_CHECK(plan.remote_block_count <= plan.blocks.size() - plan.first_remote_block); + + 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); + DORIS_CHECK(plan.first_remote_block < plan.blocks.size()); + 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; + DORIS_CHECK(remote_left <= remote_right); + const size_t remote_size = remote_right - remote_left + 1; + DORIS_CHECK(user_offset <= plan.user_left); + DORIS_CHECK(plan.user_right - user_offset < result.size); + + 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; + const size_t result_offset = copy_left - user_offset; + const size_t remote_buffer_offset = copy_left - remote_left; + DORIS_CHECK(copy_size <= result.size - result_offset); + DORIS_CHECK(copy_size <= remote_size - remote_buffer_offset); + memcpy(result.data + result_offset, remote_buffer->get() + remote_buffer_offset, 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) { + const int64_t submission_start_us = MonotonicMicros(); + Defer record_submission_latency {[&]() { + g_cached_remote_reader_async_write_submission_latency + << (MonotonicMicros() - submission_start_us); + }}; + auto* manager = _cache->async_write_manager(); + auto* inflight_index = _cache->inflight_write_buffer_index(); + DORIS_CHECK(manager != nullptr); + DORIS_CHECK(inflight_index != nullptr); + DORIS_CHECK(remote_buffer != nullptr); + + CacheContext cache_context(io_ctx); + const CacheAdmissionContext admission_context = + CacheAdmissionContext::from_cache_context(cache_context, _tablet_id); + DORIS_CHECK(plan.remote_block_count > 0); + DORIS_CHECK(plan.first_remote_block < plan.blocks.size()); + 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; + const size_t cache_block_size = static_cast<size_t>(config::file_cache_each_block_size); + DORIS_CHECK(cache_block_size > 0); + for (size_t index = plan.first_remote_block; index < remote_end; ++index) { + const auto& read_block = plan.blocks[index]; + if (!read_block.submit_write) { + continue; + } + DORIS_CHECK(read_block.source == AsyncReadBlock::Source::REMOTE); + DORIS_CHECK(read_block.range.left % cache_block_size == 0); + DORIS_CHECK(read_block.range.size() <= cache_block_size); + DORIS_CHECK(read_block.range.left >= remote_left); + DORIS_CHECK(read_block.range.right <= remote_right); + if (!manager->check_write_epoch(plan.write_epoch)) { + ++stats.async_cache_write_drop_stale_epoch; + continue; + } + + AsyncCacheWriteBufferPtr tracked_buffer; + Status status = manager->allocate_tracked_buffer(cache_block_size, &tracked_buffer); + if (!status.ok()) { + ++stats.async_cache_write_buffer_alloc_fail; + continue; + } + DORIS_CHECK(tracked_buffer->size() == cache_block_size); + DORIS_CHECK(read_block.range.size() <= tracked_buffer->size()); + const size_t remote_buffer_offset = read_block.range.left - remote_left; + DORIS_CHECK(remote_buffer_offset <= remote_size); + DORIS_CHECK(read_block.range.size() <= remote_size - remote_buffer_offset); + memcpy(tracked_buffer->data(), remote_buffer.get() + remote_buffer_offset, + read_block.range.size()); + + AsyncCacheWriteTask task { + .cache_hash = _cache_hash, + .file_offset = read_block.range.left, + .write_size = read_block.range.size(), + .buffer = tracked_buffer, + .admission_ctx = admission_context, + .submit_ts_us = MonotonicMicros(), + .write_epoch = plan.write_epoch, + .on_finalized = nullptr, + }; + std::shared_ptr<InflightWriteBufferEntry> entry; + if (config::enable_async_file_cache_write_inflight_write_buffer_index) { + entry = std::make_shared<InflightWriteBufferEntry>( + tracked_buffer, read_block.range.left, read_block.range.size(), + task.submit_ts_us); + TEST_SYNC_POINT_CALLBACK( + "CachedRemoteFileReader::_submit_async_write_tasks:before_inflight_insert", + &task); + auto existing = + inflight_index->insert_if_absent(_cache_hash, read_block.range.left, entry); Review Comment: [P1] Arm rollback as soon as this inflight owner is published. After `insert_if_absent()` succeeds, the index retains `entry`, but cleanup is only installed by the `std::function` assignment below and by the false-return path after `try_submit()`. Constructing that capturing callback may allocate, as may the deque insertion in `try_submit()`; either exception skips both cleanup paths, escapes this best-effort write after the remote read succeeded, and leaves the tracked buffer permanently indexed as inflight. Install a scope rollback immediately after successful insertion and cancel it only once admission owns guaranteed finalization (while keeping admission exceptions out of the foreground read), and inject both post-insert and queue-allocation failures. -- 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]
