github-actions[bot] commented on code in PR #65658: URL: https://github.com/apache/doris/pull/65658#discussion_r3802860579
########## be/src/io/cache/async_cache_write_manager.cpp: ########## @@ -0,0 +1,779 @@ +// 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>( Review Comment: [P1] Fully construct the token before taking this shard lock. There are two throwing ownership steps after the raw token exists: `shared_ptr` control-block allocation here can delete the raw token on failure, and the later `tokens.emplace()` can destroy the only strong owner while unwinding. In either case `AsyncCacheWriteEpochToken::~AsyncCacheWriteEpochToken()` calls `registry->release()` before this earlier `lock_guard` is destroyed, and `release()` tries to lock the same non-recursive shard mutex, permanently wedging the caller. Construct the complete owner outside the locked publication phase (or make unpublished destruction non-reentrant), and fault-inject both allocations. ########## be/src/io/cache/async_cache_write_manager.cpp: ########## @@ -0,0 +1,779 @@ +// 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); Review Comment: [P2] Make submitter registration atomic with stopping admission. These acquire/release operations are on two independent atomics, so a concurrent submitter may publish `_active_submitters=1` but still read the old `_accepting=true`, while `shutdown()` publishes false but reads the old submitter count of zero. Shutdown can then drain and stop every worker before this call takes `_queue_mutex`; this call subsequently enqueues and returns true with nobody left to activate or finalize the task, violating the documented drain/reject contract. Serialize the handshake under one lock/protocol (or otherwise forbid the crossed observation), and test a submitter that races registration itself rather than one paused after registration. ########## be/src/io/cache/cached_remote_file_reader_async_write.cpp: ########## @@ -0,0 +1,647 @@ +// 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/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"); + +} // 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) { + 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; + } + } + 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; + return false; + } + ++stats.block_wait_success; + } + 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); Review Comment: [P2] Add a lifecycle-safe promotion from a successful async cache hit into the Doris reader's direct map. A reader constructed while the cache is cold snapshots an empty `_cache_file_readers`; the synchronous path repairs that map for existing DOWNLOADED blocks and newly finalized misses, but this path only touches LRU state. No manager/probe callback can update the reader, so under stable async mode later reads keep doing inflight lookup, plan allocation, and the cache-wide probe lock instead of returning to the default direct-read fast path. The promotion must atomically verify the cell is still current/non-deleting and handle a stale same-offset reader entry—blindly calling `_insert_file_reader()` while the probe owns a reference creates the cleanup problem described separately. Add a cold-reader test with direct reads enabled; current async tests and the benchmark disable them. ########## be/src/io/cache/file_block.cpp: ########## @@ -331,36 +331,50 @@ std::string FileBlock::get_cache_file() const { return _mgr->_storage->get_local_file(this->_key); } -FileBlocksHolder::~FileBlocksHolder() { - for (auto file_block_it = file_blocks.begin(); file_block_it != file_blocks.end();) { - auto current_file_block_it = file_block_it; - auto& file_block = *current_file_block_it; - BlockFileCache* _mgr = file_block->_mgr; +void FileBlock::release_cache_reference(std::shared_ptr<FileBlock>& file_block, + CacheReferenceRole role) { + if (!file_block) { + return; + } + BlockFileCache* mgr = file_block->_mgr; + { + bool should_remove = false; { - bool should_remove = false; - { - std::lock_guard block_lock(file_block->_mutex); + std::lock_guard block_lock(file_block->_mutex); + if (role == CacheReferenceRole::HOLDER) { file_block->complete_unlocked(block_lock); - if (file_block.use_count() == 2 && - (file_block->is_deleting() || - file_block->state_unlock(block_lock) == FileBlock::State::EMPTY)) { - should_remove = true; - } } - if (should_remove) { - SCOPED_CACHE_LOCK(_mgr->_mutex, _mgr); - std::lock_guard block_lock(file_block->_mutex); - if (file_block.use_count() == 2) { - DCHECK(file_block->state_unlock(block_lock) != FileBlock::State::DOWNLOADING); - // one in cache, one in here - if (file_block->is_deleting() || - file_block->state_unlock(block_lock) == FileBlock::State::EMPTY) { - _mgr->remove(file_block, cache_lock, block_lock, false); - } + if (file_block.use_count() == 2 && Review Comment: [P2] Do not require exactly two owners before arranging deleting-block cleanup. With default direct reads, a warm reader already owns this DOWNLOADED block beside the cache; async `probe()` adds a third reference. If the backing file is missing, self-heal marks the cell deleting, but this check sees three and skips removal; the following `reset()` leaves cache + reader at two with no later cleanup event. Subsequent workers get the same deleting cell and skip it, so the hash keeps falling back remotely and cannot refill until unrelated cleanup occurs. Make deferred removal account for probe/holder and direct-reader ownership instead of sampling an exact pre-reset count, and run the missing-file self-heal test with direct reads enabled. -- 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]
