This is an automated email from the ASF dual-hosted git repository. Gabriel39 pushed a commit to branch 4.1.5-iceberg-opt in repository https://gitbox.apache.org/repos/asf/doris.git
commit 94757e1743e9e14858dfed39e1c1629649ed18fb Author: Gabriel <[email protected]> AuthorDate: Wed Oct 28 12:58:19 2026 +0800 [improvement](file-cache) adapt cached external reads to I/O latency A fully warm external scan can stall on a congested cache disk even when remote storage can serve the same immutable bytes faster. Add an optional, cache-local controller that compares measured per-byte costs within bounded resource/read-size shards. Each side retains 50 recent samples; an idle window expires after 30 seconds. Require 50 local observations and a 5 ms local latency floor before considering bypass. Real remote bootstrap probes and periodic local recovery probes avoid permanently trusting stale estimates. Route fully downloaded disk-cache hits before the sync/async dispatch, so both write modes share the policy. Mixed misses and inflight memory payloads retain the existing planner. Bypass reads the requested remote range directly, retains cached blocks, and never takes downloader ownership or writes them again. On remote errors or short reads, overwrite the complete range from local cache; propagate cancellation and retain normal cache-file self-healing. A move-only permit bounds extra remote requests to 8 per cache, with an 8 MiB request limit by default. The feature is disabled by default and controlled by enable_file_cache_io_adaptor and validated mutable tuning options. Add 12 Profile counters for lookup cost, local/remote service time, attempted and returned bypass bytes, bootstrap/recovery probes, budget rejections and local fallback. Propagate them through per-read statistics, merges, delta calculation and Profile reporting. Measured effect in the opt-in, steady-state FileCache-path benchmark: - Five alternating on/off rounds, 200 x 64 KiB reads per group. Windows are warmed with 50 actual reads of each path. The remote reader is file-backed; service delays are injected, so these results measure routing behavior. - Local 8 ms / remote 1 ms: median 1,624.026 ms disabled versus 256.277 ms enabled, a 6.34x speedup / 84.22% latency reduction. Each enabled group has 194 bypasses (12,713,984 bytes) and 6 local recovery probes. - Local 1 ms / remote 8 ms: median 214.784 ms disabled versus 214.827 ms enabled (+0.02%), with zero remote bypasses. - Both scenarios preserve every returned byte and perform zero cache writes. Production query gains, including end-to-end runtime-filter timing, require separate measurement. Validation: - BE Release build and GLIBC 2.17 symbol compatibility check passed. - 41 focused BE tests passed: 15 controller tests, 8 reader integration tests, 13 existing async-reader tests and 5 statistics/Profile tests. - All 15 controller tests passed under ASan and UBSan, including concurrent admission and move-only permit release. - Both opt-in reader benchmarks passed. Integration coverage includes both write modes, cross-block/EOF ranges, short/error/cancelled remote reads, missing cache files, mixed misses and inflight-memory hits. - clang-format 16 and git diff --check passed for all changed C++ files. --- be/src/common/config.cpp | 23 ++ be/src/common/config.h | 10 + be/src/io/cache/block_file_cache.cpp | 2 + be/src/io/cache/block_file_cache.h | 3 + be/src/io/cache/block_file_cache_profile.cpp | 48 +++ be/src/io/cache/block_file_cache_profile.h | 12 + be/src/io/cache/cached_remote_file_reader.cpp | 155 +++++++- be/src/io/cache/cached_remote_file_reader.h | 7 + be/src/io/cache/file_block.cpp | 1 + be/src/io/cache/file_cache_common.h | 13 + be/src/io/cache/file_cache_io_adaptor.cpp | 124 ++++++ be/src/io/cache/file_cache_io_adaptor.h | 117 ++++++ be/src/io/io_common.h | 24 ++ .../block_file_cache_profile_reporter_test.cpp | 48 +++ .../io/cache/cached_remote_file_reader_test.cpp | 432 +++++++++++++++++++++ be/test/io/cache/file_cache_io_adaptor_test.cpp | 182 +++++++++ 16 files changed, 1200 insertions(+), 1 deletion(-) diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp index 0e34b88d708..285443ee04b 100644 --- a/be/src/common/config.cpp +++ b/be/src/common/config.cpp @@ -1322,6 +1322,29 @@ DEFINE_mBool(file_cache_enable_only_warm_up_idx, "false"); DEFINE_Int32(file_cache_downloader_thread_num_min, "32"); DEFINE_Int32(file_cache_downloader_thread_num_max, "32"); +// Latency-based routing for fully cached external data reads. Bypass never invalidates or +// rewrites an existing cache block. Keep disabled until the storage environment is measured. +DEFINE_mBool(enable_file_cache_io_adaptor, "false"); +DEFINE_mInt64(file_cache_io_adaptor_min_samples, "50"); +DEFINE_mInt64(file_cache_io_adaptor_min_local_latency_us, "5000"); +DEFINE_mDouble(file_cache_io_adaptor_skip_read_factor, "1.0"); +DEFINE_mInt64(file_cache_io_adaptor_probe_interval, "32"); +DEFINE_mInt64(file_cache_io_adaptor_sample_ttl_ms, "30000"); +DEFINE_mInt32(file_cache_io_adaptor_max_remote_inflight, "8"); +DEFINE_mInt64(file_cache_io_adaptor_max_request_bytes, "8388608"); +DEFINE_Validator(file_cache_io_adaptor_min_samples, [](int64_t v) { return v > 0 && v <= 50; }); +DEFINE_Validator(file_cache_io_adaptor_min_local_latency_us, + [](int64_t v) { return v >= 0 && v <= 10000000; }); +DEFINE_Validator(file_cache_io_adaptor_skip_read_factor, + [](double v) { return v > 0 && v <= 100; }); +DEFINE_Validator(file_cache_io_adaptor_probe_interval, [](int64_t v) { return v > 1; }); +DEFINE_Validator(file_cache_io_adaptor_sample_ttl_ms, + [](int64_t v) { return v > 0 && v <= 3600000; }); +DEFINE_Validator(file_cache_io_adaptor_max_remote_inflight, + [](int32_t v) { return v > 0 && v <= 1024; }); +DEFINE_Validator(file_cache_io_adaptor_max_request_bytes, + [](int64_t v) { return v > 0 && v <= 67108864; }); + // async file cache write DEFINE_mBool(enable_async_file_cache_write, "false"); DEFINE_mInt32(async_file_cache_write_workers_per_disk, "16"); diff --git a/be/src/common/config.h b/be/src/common/config.h index 3931b11efed..7ad17b21abe 100644 --- a/be/src/common/config.h +++ b/be/src/common/config.h @@ -1349,6 +1349,16 @@ DECLARE_mBool(enable_evaluate_shadow_queue_diff); DECLARE_mBool(file_cache_enable_only_warm_up_idx); +// Adaptive external file-cache reads +DECLARE_mBool(enable_file_cache_io_adaptor); +DECLARE_mInt64(file_cache_io_adaptor_min_samples); +DECLARE_mInt64(file_cache_io_adaptor_min_local_latency_us); +DECLARE_mDouble(file_cache_io_adaptor_skip_read_factor); +DECLARE_mInt64(file_cache_io_adaptor_probe_interval); +DECLARE_mInt64(file_cache_io_adaptor_sample_ttl_ms); +DECLARE_mInt32(file_cache_io_adaptor_max_remote_inflight); +DECLARE_mInt64(file_cache_io_adaptor_max_request_bytes); + // async file cache write DECLARE_mBool(enable_async_file_cache_write); DECLARE_mInt32(async_file_cache_write_workers_per_disk); diff --git a/be/src/io/cache/block_file_cache.cpp b/be/src/io/cache/block_file_cache.cpp index f78fe10ef41..94eec211e1c 100644 --- a/be/src/io/cache/block_file_cache.cpp +++ b/be/src/io/cache/block_file_cache.cpp @@ -28,6 +28,7 @@ #include "common/status.h" #include "cpp/sync_point.h" +#include "io/cache/file_cache_io_adaptor.h" #include "runtime/exec_env.h" #if defined(__APPLE__) @@ -448,6 +449,7 @@ BlockFileCache::BlockFileCache(const std::string& cache_base_path, _ttl_queue = LRUQueue(cache_settings.ttl_queue_size, cache_settings.ttl_queue_elements, std::numeric_limits<int>::max()); + _io_adaptor = std::make_unique<FileCacheIOAdaptor>(); _lru_recorder = std::make_unique<LRUQueueRecorder>(this); _lru_dumper = std::make_unique<CacheLRUDumper>(this, _lru_recorder.get()); if (cache_settings.storage == "memory") { diff --git a/be/src/io/cache/block_file_cache.h b/be/src/io/cache/block_file_cache.h index caf9fed6b2a..dc00e2f0628 100644 --- a/be/src/io/cache/block_file_cache.h +++ b/be/src/io/cache/block_file_cache.h @@ -79,6 +79,7 @@ private: } \ LockScopedTimer cache_lock_timer; +class FileCacheIOAdaptor; class AsyncCacheWriteManager; class FSFileCacheStorage; class InflightWriteBufferIndex; @@ -212,6 +213,7 @@ public: // Get storage for inspection FileCacheStorage* get_storage() const { return _storage.get(); } + FileCacheIOAdaptor& io_adaptor() { return *_io_adaptor; } /** * Given an `offset` and `size` representing [offset, offset + size) bytes interval, @@ -666,6 +668,7 @@ private: // coz it will use other members of BlockFileCache // so join this async load thread first std::unique_ptr<FileCacheStorage> _storage; + std::unique_ptr<FileCacheIOAdaptor> _io_adaptor; std::shared_ptr<bvar::LatencyRecorder> _lru_dump_latency_us; std::mutex _dump_lru_queues_mtx; NeedUpdateLRUBlocks _need_update_lru_blocks; diff --git a/be/src/io/cache/block_file_cache_profile.cpp b/be/src/io/cache/block_file_cache_profile.cpp index 366cf6fd684..2135c1dc7ca 100644 --- a/be/src/io/cache/block_file_cache_profile.cpp +++ b/be/src/io/cache/block_file_cache_profile.cpp @@ -99,6 +99,18 @@ FileCacheStatistics diff_file_cache_statistics(const FileCacheStatistics& curren SUBTRACT_FIELD(lock_wait_timer); SUBTRACT_FIELD(get_timer); SUBTRACT_FIELD(set_timer); + SUBTRACT_FIELD(io_adaptor_lookup_count); + SUBTRACT_FIELD(io_adaptor_lookup_time_ns); + SUBTRACT_FIELD(io_adaptor_local_reads); + SUBTRACT_FIELD(io_adaptor_local_time_ns); + SUBTRACT_FIELD(io_adaptor_bypass_count); + SUBTRACT_FIELD(io_adaptor_bypass_bytes); + SUBTRACT_FIELD(io_adaptor_remote_bytes); + SUBTRACT_FIELD(io_adaptor_remote_time_ns); + SUBTRACT_FIELD(io_adaptor_local_probes); + SUBTRACT_FIELD(io_adaptor_remote_probes); + SUBTRACT_FIELD(io_adaptor_budget_rejected); + SUBTRACT_FIELD(io_adaptor_fallback_local); SUBTRACT_FIELD(async_cache_write_submitted); SUBTRACT_FIELD(async_cache_write_rejected); SUBTRACT_FIELD(async_cache_write_buffer_alloc_fail); @@ -176,6 +188,30 @@ FileCacheProfileReporter::FileCacheProfileReporter(RuntimeProfile* profile, lock_wait_timer = ADD_CHILD_TIMER_WITH_LEVEL(profile, "LockWaitTimer", cache_profile, 1); get_timer = ADD_CHILD_TIMER_WITH_LEVEL(profile, "GetTimer", cache_profile, 1); set_timer = ADD_CHILD_TIMER_WITH_LEVEL(profile, "SetTimer", cache_profile, 1); + io_adaptor_lookup_count = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "IOAdaptorLookupCount", + TUnit::UNIT, cache_profile, 1); + io_adaptor_lookup_time_ns = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "IOAdaptorLookupTime", + TUnit::TIME_NS, cache_profile, 1); + io_adaptor_local_reads = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "IOAdaptorLocalReadCount", + TUnit::UNIT, cache_profile, 1); + io_adaptor_local_time_ns = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "IOAdaptorLocalReadTime", + TUnit::TIME_NS, cache_profile, 1); + io_adaptor_bypass_count = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "IOAdaptorBypassCount", + TUnit::UNIT, cache_profile, 1); + io_adaptor_bypass_bytes = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "IOAdaptorBypassBytes", + TUnit::BYTES, cache_profile, 1); + io_adaptor_remote_bytes = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "IOAdaptorRemoteReadBytes", + TUnit::BYTES, cache_profile, 1); + io_adaptor_remote_time_ns = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "IOAdaptorRemoteReadTime", + TUnit::TIME_NS, cache_profile, 1); + io_adaptor_local_probes = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "IOAdaptorLocalProbeCount", + TUnit::UNIT, cache_profile, 1); + io_adaptor_remote_probes = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "IOAdaptorRemoteProbeCount", + TUnit::UNIT, cache_profile, 1); + io_adaptor_budget_rejected = ADD_CHILD_COUNTER_WITH_LEVEL( + profile, "IOAdaptorBudgetRejectedCount", TUnit::UNIT, cache_profile, 1); + io_adaptor_fallback_local = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "IOAdaptorFallbackLocalCount", + TUnit::UNIT, cache_profile, 1); async_cache_write_submitted = ADD_CHILD_COUNTER_WITH_LEVEL( profile, "AsyncCacheWriteSubmittedCount", TUnit::UNIT, cache_profile, 1); async_cache_write_rejected = ADD_CHILD_COUNTER_WITH_LEVEL( @@ -298,6 +334,18 @@ void FileCacheProfileReporter::update(const FileCacheStatistics* statistics) con COUNTER_UPDATE(lock_wait_timer, statistics->lock_wait_timer); COUNTER_UPDATE(get_timer, statistics->get_timer); COUNTER_UPDATE(set_timer, statistics->set_timer); + COUNTER_UPDATE(io_adaptor_lookup_count, statistics->io_adaptor_lookup_count); + COUNTER_UPDATE(io_adaptor_lookup_time_ns, statistics->io_adaptor_lookup_time_ns); + COUNTER_UPDATE(io_adaptor_local_reads, statistics->io_adaptor_local_reads); + COUNTER_UPDATE(io_adaptor_local_time_ns, statistics->io_adaptor_local_time_ns); + COUNTER_UPDATE(io_adaptor_bypass_count, statistics->io_adaptor_bypass_count); + COUNTER_UPDATE(io_adaptor_bypass_bytes, statistics->io_adaptor_bypass_bytes); + COUNTER_UPDATE(io_adaptor_remote_bytes, statistics->io_adaptor_remote_bytes); + COUNTER_UPDATE(io_adaptor_remote_time_ns, statistics->io_adaptor_remote_time_ns); + COUNTER_UPDATE(io_adaptor_local_probes, statistics->io_adaptor_local_probes); + COUNTER_UPDATE(io_adaptor_remote_probes, statistics->io_adaptor_remote_probes); + COUNTER_UPDATE(io_adaptor_budget_rejected, statistics->io_adaptor_budget_rejected); + COUNTER_UPDATE(io_adaptor_fallback_local, statistics->io_adaptor_fallback_local); COUNTER_UPDATE(async_cache_write_submitted, statistics->async_cache_write_submitted); COUNTER_UPDATE(async_cache_write_rejected, statistics->async_cache_write_rejected); COUNTER_UPDATE(async_cache_write_buffer_alloc_fail, diff --git a/be/src/io/cache/block_file_cache_profile.h b/be/src/io/cache/block_file_cache_profile.h index 4ffe8f0835e..f5015999ce4 100644 --- a/be/src/io/cache/block_file_cache_profile.h +++ b/be/src/io/cache/block_file_cache_profile.h @@ -90,6 +90,18 @@ struct FileCacheProfileReporter { RuntimeProfile::Counter* lock_wait_timer = nullptr; RuntimeProfile::Counter* get_timer = nullptr; RuntimeProfile::Counter* set_timer = nullptr; + RuntimeProfile::Counter* io_adaptor_lookup_count = nullptr; + RuntimeProfile::Counter* io_adaptor_lookup_time_ns = nullptr; + RuntimeProfile::Counter* io_adaptor_local_reads = nullptr; + RuntimeProfile::Counter* io_adaptor_local_time_ns = nullptr; + RuntimeProfile::Counter* io_adaptor_bypass_count = nullptr; + RuntimeProfile::Counter* io_adaptor_bypass_bytes = nullptr; + RuntimeProfile::Counter* io_adaptor_remote_bytes = nullptr; + RuntimeProfile::Counter* io_adaptor_remote_time_ns = nullptr; + RuntimeProfile::Counter* io_adaptor_local_probes = nullptr; + RuntimeProfile::Counter* io_adaptor_remote_probes = nullptr; + RuntimeProfile::Counter* io_adaptor_budget_rejected = nullptr; + RuntimeProfile::Counter* io_adaptor_fallback_local = nullptr; RuntimeProfile::Counter* async_cache_write_submitted = nullptr; RuntimeProfile::Counter* async_cache_write_rejected = nullptr; RuntimeProfile::Counter* async_cache_write_buffer_alloc_fail = nullptr; diff --git a/be/src/io/cache/cached_remote_file_reader.cpp b/be/src/io/cache/cached_remote_file_reader.cpp index 0d0a0dbf3cb..fa54ee1a33e 100644 --- a/be/src/io/cache/cached_remote_file_reader.cpp +++ b/be/src/io/cache/cached_remote_file_reader.cpp @@ -48,6 +48,7 @@ #include "io/cache/block_file_cache_factory.h" #include "io/cache/block_file_cache_profile.h" #include "io/cache/file_block.h" +#include "io/cache/file_cache_io_adaptor.h" #include "io/cache/peer_file_cache_reader.h" #include "io/cache/remote_scan_cache_write_limiter.h" #include "io/fs/file_reader.h" @@ -116,6 +117,19 @@ static bool use_remote_only_on_cache_miss(const IOContext* io_ctx) { return limiter != nullptr && limiter->remote_only_on_miss(); } +namespace { +FileCacheIOAdaptor::Options file_cache_io_adaptor_options() { + FileCacheIOAdaptor::Options options; + options.min_samples = static_cast<size_t>(config::file_cache_io_adaptor_min_samples); + options.min_local_latency_ns = config::file_cache_io_adaptor_min_local_latency_us * 1000; + options.skip_read_factor = config::file_cache_io_adaptor_skip_read_factor; + options.probe_interval = static_cast<uint64_t>(config::file_cache_io_adaptor_probe_interval); + options.sample_ttl_ns = config::file_cache_io_adaptor_sample_ttl_ms * 1'000'000; + options.max_remote_inflight = config::file_cache_io_adaptor_max_remote_inflight; + return options; +} +} // namespace + CachedRemoteFileReader::CachedRemoteFileReader(FileReaderSPtr remote_file_reader, const FileReaderOptions& opts) : _is_doris_table(opts.is_doris_table), @@ -124,6 +138,13 @@ CachedRemoteFileReader::CachedRemoteFileReader(FileReaderSPtr remote_file_reader _tablet_id(opts.is_doris_table ? opts.tablet_id : 0), _storage_resource_id(opts.storage_resource_id), _remote_file_reader(std::move(remote_file_reader)) { + const auto uri = std::string_view(path().native()); + const auto scheme = uri.find("://"); + const auto authority_end = scheme == std::string_view::npos ? 0 : uri.find('/', scheme + 3); + _io_adaptor_resource_key = + _storage_resource_id.empty() + ? std::hash<std::string_view> {}(uri.substr(0, authority_end)) + : std::hash<std::string> {}(_storage_resource_id); DCHECK(!_is_doris_table || _tablet_id > 0); if (_is_doris_table) { _init_doris_table_cache(); @@ -328,7 +349,11 @@ Status execute_s3_read(size_t empty_start, size_t& size, std::unique_ptr<char[]> s3_read_counter << 1; SCOPED_RAW_TIMER(&stats.remote_read_timer); stats.from_peer_cache = false; - return remote_file_reader->read_at(empty_start, Slice(buffer.get(), size), &size, io_ctx); + auto st = remote_file_reader->read_at(empty_start, Slice(buffer.get(), size), &size, io_ctx); + if (st.ok()) { + stats.remote_io_sample_bytes += size; + } + return st; } CloudWarmUpManager& get_warm_up_manager() { @@ -1039,6 +1064,7 @@ Status CachedRemoteFileReader::_read_remaining_blocks_from_cache( RETURN_IF_ERROR(_remote_file_reader->read_at( current_offset, Slice(result.data + (current_offset - offset), read_size), &remote_bytes_read, io_ctx)); + stats.remote_io_sample_bytes += remote_bytes_read; indirect_read_bytes += read_size; source_read_breakdown.remote_bytes += remote_bytes_read; DCHECK(remote_bytes_read == read_size); @@ -1113,6 +1139,7 @@ Status CachedRemoteFileReader::_read_remote_only_on_cache_miss( SCOPED_RAW_TIMER(&stats.remote_read_timer); RETURN_IF_ERROR(_remote_file_reader->read_at(offset, Slice(result.data, bytes_req), &remote_bytes_read, io_ctx)); + stats.remote_io_sample_bytes += remote_bytes_read; *bytes_read = remote_bytes_read; DCHECK_EQ(*bytes_read, bytes_req); source_read_breakdown.remote_bytes += remote_bytes_read; @@ -1180,6 +1207,105 @@ Status CachedRemoteFileReader::_read_remote_only_on_cache_miss( return Status::OK(); } +bool CachedRemoteFileReader::_io_adaptor_eligible(size_t bytes, const IOContext* io_ctx) const { + // Query data reads can bypass a slow disk; index reads, warmup and inflight memory retain + // their existing policies. Bound both remote request count and bytes in flight. + return config::enable_file_cache_io_adaptor && !_is_doris_table && !io_ctx->is_dryrun && + !io_ctx->is_warmup && !io_ctx->is_index_data && !io_ctx->is_inverted_index && + !io_ctx->should_stop && io_ctx->read_file_cache && bytes > 0 && + bytes <= static_cast<size_t>(config::file_cache_io_adaptor_max_request_bytes) && + _cache->get_storage()->get_type() == FileCacheStorageType::DISK; +} + +Status CachedRemoteFileReader::_try_read_with_io_adaptor( + size_t offset, Slice result, size_t bytes_req, size_t* bytes_read, const IOContext* io_ctx, + ReadStatistics& stats, SourceReadBreakdown& sources, bool* handled) { + *handled = false; + FileBlocks blocks; + bool fully_covered = false; + CacheContext context(io_ctx); + context.stats = &stats; + context.tablet_id = _tablet_id; + ++stats.io_adaptor_lookup_count; + { + SCOPED_RAW_TIMER(&stats.io_adaptor_lookup_time_ns); + RETURN_IF_ERROR(_cache->get_downloaded_blocks_if_fully_covered( + _cache_hash, offset, bytes_req, context, &blocks, &fully_covered)); + } + if (!fully_covered) { + // A mixed hit/miss or inflight-buffer request follows the existing sync/async planner. + // In particular, a bypass must never acquire downloader ownership or cause writeback. + return Status::OK(); + } + auto options = file_cache_io_adaptor_options(); + auto& adaptor = _cache->io_adaptor(); + auto decision = adaptor.select(_io_adaptor_resource_key, bytes_req, MonotonicNanos(), options); + using Route = FileCacheIOAdaptor::Route; + stats.io_adaptor_local_probes += decision.route == Route::LOCAL_PROBE; + stats.io_adaptor_remote_probes += decision.route == Route::REMOTE_PROBE; + stats.io_adaptor_budget_rejected += decision.route == Route::LOCAL_BUDGET; + if (decision.read_remote()) { + // Requested and returned bytes differ for failed/short reads recovered from local cache. + ++stats.io_adaptor_bypass_count; + stats.io_adaptor_bypass_bytes += bytes_req; + size_t remote_bytes = 0; + s3_read_counter << 1; + const auto start = MonotonicNanos(); + Status status = _remote_file_reader->read_at(offset, Slice(result.data, bytes_req), + &remote_bytes, io_ctx); + const auto elapsed = MonotonicNanos() - start; + stats.io_adaptor_remote_time_ns += elapsed; + stats.remote_read_timer += elapsed; + stats.io_adaptor_remote_bytes += remote_bytes; + if (status.ok() && remote_bytes == bytes_req) { + adaptor.record_remote(_io_adaptor_resource_key, remote_bytes, elapsed, MonotonicNanos(), + options); + sources.remote_bytes += remote_bytes; + stats.hit_cache = false; + stats.skip_cache = true; + *bytes_read = remote_bytes; + *handled = true; + return Status::OK(); + } + adaptor.record_remote_failure(_io_adaptor_resource_key, bytes_req, MonotonicNanos(), + options); + if (io_ctx->should_stop || status.is<ErrorCode::CANCELLED>()) { + return status.ok() ? Status::Cancelled("Adaptive cache read cancelled") : status; + } + // The same immutable range is still cached. Overwrite the complete caller range after + // an error/short remote read; never publish the partially overwritten buffer as success. + ++stats.io_adaptor_fallback_local; + } + const auto local_start = MonotonicNanos(); + ++stats.io_adaptor_local_reads; + Defer local_timer {[&] { + const auto elapsed = MonotonicNanos() - local_start; + stats.io_adaptor_local_time_ns += elapsed; + stats.local_read_timer += elapsed; + }}; + for (const auto& block : blocks) { + const auto left = std::max(offset, block->range().left); + const auto right = std::min(offset + bytes_req - 1, block->range().right); + Status status = block->read(Slice(result.data + left - offset, right - left + 1), + left - block->range().left); + if (!status.ok()) { + // Concurrent eviction can remove a previously downloaded file. The ordinary path + // already handles self-healing and remote fallback, so retry the full request there. + if (status.is<ErrorCode::NOT_FOUND>()) { + _cache->remove_if_cached_async(_cache_hash); + } + return Status::OK(); + } + _cache->touch_probe_block_if_cached(block, context); + } + const auto elapsed = MonotonicNanos() - local_start; + adaptor.record_local(_io_adaptor_resource_key, bytes_req, elapsed, MonotonicNanos(), options); + sources.local_bytes += bytes_req; + *bytes_read = bytes_req; + *handled = true; + return Status::OK(); +} + Status CachedRemoteFileReader::read_at_impl(size_t offset, Slice result, size_t* bytes_read, const IOContext* io_ctx) { SCOPED_CONCURRENCY_COUNT(ConcurrencyStatsManager::instance().cached_remote_reader_read_at); @@ -1222,6 +1348,13 @@ Status CachedRemoteFileReader::read_at_impl(size_t offset, Slice result, size_t* path().native(), offset, bytes_req, read_at_sw.elapsed_time_milliseconds(), io_ctx->is_warmup); } + if (read_st.ok() && _io_adaptor_eligible(bytes_req, io_ctx) && + stats.remote_io_sample_bytes > 0 && stats.io_adaptor_bypass_count == 0 && + !stats.from_peer_cache) { + _cache->io_adaptor().record_remote( + _io_adaptor_resource_key, static_cast<size_t>(stats.remote_io_sample_bytes), + stats.remote_read_timer, MonotonicNanos(), file_cache_io_adaptor_options()); + } if (read_st.ok() && !is_dryrun) { // Only successful reads contribute to query profile and file-cache metrics. const auto file_cache_read_type = @@ -1250,6 +1383,14 @@ Status CachedRemoteFileReader::read_at_impl(size_t offset, Slice result, size_t* } }}; + if (_io_adaptor_eligible(bytes_req, io_ctx)) { + bool handled = false; + read_st = _try_read_with_io_adaptor(offset, result, bytes_req, bytes_read, io_ctx, stats, + source_read_breakdown, &handled); + if (!read_st.ok() || handled) { + return read_st; + } + } if (use_remote_only_on_cache_miss(io_ctx)) { read_st = _read_remote_only_on_cache_miss(offset, result, bytes_req, is_dryrun, bytes_read, stats, source_read_breakdown, io_ctx); @@ -1380,6 +1521,18 @@ void CachedRemoteFileReader::_update_stats(const ReadStatistics& read_stats, statis->lock_wait_timer += read_stats.lock_wait_timer; statis->get_timer += read_stats.get_timer; statis->set_timer += read_stats.set_timer; + statis->io_adaptor_lookup_count += read_stats.io_adaptor_lookup_count; + statis->io_adaptor_lookup_time_ns += read_stats.io_adaptor_lookup_time_ns; + statis->io_adaptor_local_reads += read_stats.io_adaptor_local_reads; + statis->io_adaptor_local_time_ns += read_stats.io_adaptor_local_time_ns; + statis->io_adaptor_bypass_count += read_stats.io_adaptor_bypass_count; + statis->io_adaptor_bypass_bytes += read_stats.io_adaptor_bypass_bytes; + statis->io_adaptor_remote_bytes += read_stats.io_adaptor_remote_bytes; + statis->io_adaptor_remote_time_ns += read_stats.io_adaptor_remote_time_ns; + statis->io_adaptor_local_probes += read_stats.io_adaptor_local_probes; + statis->io_adaptor_remote_probes += read_stats.io_adaptor_remote_probes; + statis->io_adaptor_budget_rejected += read_stats.io_adaptor_budget_rejected; + statis->io_adaptor_fallback_local += read_stats.io_adaptor_fallback_local; statis->async_cache_write_submitted += read_stats.async_cache_write_submitted; statis->async_cache_write_rejected += read_stats.async_cache_write_rejected; statis->async_cache_write_buffer_alloc_fail += read_stats.async_cache_write_buffer_alloc_fail; diff --git a/be/src/io/cache/cached_remote_file_reader.h b/be/src/io/cache/cached_remote_file_reader.h index 68b112d9ece..222082da234 100644 --- a/be/src/io/cache/cached_remote_file_reader.h +++ b/be/src/io/cache/cached_remote_file_reader.h @@ -426,6 +426,13 @@ private: const SourceReadBreakdown& source_read_breakdown, FileCacheStatistics* state, FileCacheReadType read_type) const; + bool _io_adaptor_eligible(size_t bytes, const IOContext* io_ctx) const; + Status _try_read_with_io_adaptor(size_t offset, Slice result, size_t bytes_req, + size_t* bytes_read, const IOContext* io_ctx, + ReadStatistics& stats, SourceReadBreakdown& sources, + bool* handled); + uint64_t _io_adaptor_resource_key = 0; + bool _is_doris_table = false; CacheAlignMode _cache_align_mode {CacheAlignMode::ALIGN_TO_BLOCK}; CacheWriteMode _cache_write_mode {CacheWriteMode::DEFAULT}; diff --git a/be/src/io/cache/file_block.cpp b/be/src/io/cache/file_block.cpp index d9d57b08bf9..f221d9eb932 100644 --- a/be/src/io/cache/file_block.cpp +++ b/be/src/io/cache/file_block.cpp @@ -230,6 +230,7 @@ Status FileBlock::finalize() { } Status FileBlock::read(Slice buffer, size_t read_offset) { + TEST_SYNC_POINT("FileBlock::adaptor_benchmark_read"); return _mgr->_storage->read(_key, read_offset, buffer); } diff --git a/be/src/io/cache/file_cache_common.h b/be/src/io/cache/file_cache_common.h index 554d049b21e..5b7aa5be793 100644 --- a/be/src/io/cache/file_cache_common.h +++ b/be/src/io/cache/file_cache_common.h @@ -78,6 +78,7 @@ struct ReadStatistics { int64_t bytes_read_from_peer = 0; int64_t bytes_write_into_file_cache = 0; int64_t remote_read_timer = 0; + int64_t remote_io_sample_bytes = 0; int64_t peer_read_timer = 0; int64_t remote_wait_timer = 0; // wait for other downloader int64_t local_read_timer = 0; @@ -87,6 +88,18 @@ struct ReadStatistics { int64_t lock_wait_timer = 0; int64_t get_timer = 0; int64_t set_timer = 0; + int64_t io_adaptor_lookup_count = 0; + int64_t io_adaptor_lookup_time_ns = 0; + int64_t io_adaptor_local_reads = 0; + int64_t io_adaptor_local_time_ns = 0; + int64_t io_adaptor_bypass_count = 0; + int64_t io_adaptor_bypass_bytes = 0; + int64_t io_adaptor_remote_bytes = 0; + int64_t io_adaptor_remote_time_ns = 0; + int64_t io_adaptor_local_probes = 0; + int64_t io_adaptor_remote_probes = 0; + int64_t io_adaptor_budget_rejected = 0; + int64_t io_adaptor_fallback_local = 0; int64_t async_cache_write_submitted = 0; int64_t async_cache_write_rejected = 0; int64_t async_cache_write_buffer_alloc_fail = 0; diff --git a/be/src/io/cache/file_cache_io_adaptor.cpp b/be/src/io/cache/file_cache_io_adaptor.cpp new file mode 100644 index 00000000000..16e66b4e6e6 --- /dev/null +++ b/be/src/io/cache/file_cache_io_adaptor.cpp @@ -0,0 +1,124 @@ +// 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/file_cache_io_adaptor.h" + +#include <cassert> + +namespace doris::io { +void FileCacheIOAdaptor::Window::expire(int64_t now_ns, int64_t ttl_ns) { + // An idle window must not reuse estimates from an earlier workload. + if (count != 0 && now_ns - last_ns >= ttl_ns) { + *this = Window {}; + } +} + +void FileCacheIOAdaptor::Window::record(size_t bytes, int64_t elapsed_ns, int64_t now_ns) { + assert(bytes != 0 && elapsed_ns >= 0); + if (count == WINDOW_SIZE) { + sum_ns -= samples[next].latency_ns; + sum_ns_per_byte -= samples[next].ns_per_byte; + } else { + ++count; + } + samples[next] = {elapsed_ns, static_cast<double>(elapsed_ns) / static_cast<double>(bytes)}; + sum_ns += elapsed_ns; + sum_ns_per_byte += samples[next].ns_per_byte; + next = (next + 1) % WINDOW_SIZE; + last_ns = now_ns; +} + +size_t FileCacheIOAdaptor::bucket_index(uint64_t resource, size_t bytes) { + size_t size_bucket = 0; + // Separate tiny metadata reads from large merged ranges. Hash collisions only share + // performance estimates; they never share cache identities or downloaded data. + for (size_t upper = 16 * 1024; bytes > upper && size_bucket + 1 < SIZE_BUCKETS; upper *= 4) { + ++size_bucket; + } + return (resource % RESOURCE_SHARDS) * SIZE_BUCKETS + size_bucket; +} + +FileCacheIOAdaptor::Decision FileCacheIOAdaptor::select(uint64_t resource, size_t bytes, + int64_t now_ns, const Options& options) { + assert(options.min_samples > 0 && options.min_samples <= WINDOW_SIZE); + assert(options.probe_interval > 0 && options.max_remote_inflight > 0); + auto& bucket = _buckets[bucket_index(resource, bytes)]; + std::lock_guard lock(bucket.mutex); + auto& local = bucket.local; + auto& remote = bucket.remote; + local.expire(now_ns, options.sample_ttl_ns); + remote.expire(now_ns, options.sample_ttl_ns); + Decision decision; + decision.local_latency_ns = local.count == 0 ? 0 : local.sum_ns / int64_t(local.count); + decision.remote_latency_ns = remote.count == 0 ? 0 : remote.sum_ns / int64_t(remote.count); + if (now_ns < bucket.remote_failure_until_ns || local.count < options.min_samples || + decision.local_latency_ns < options.min_local_latency_ns) { + return decision; + } + ++bucket.decisions; + if (remote.count < options.min_samples) { + // A fully warm cache produces no natural misses. Bounded real remote probes bootstrap + // its estimate without treating an unmeasured network as faster than the disk. + if (bucket.decisions % options.probe_interval != 0) { + return decision; + } + decision.route = Route::REMOTE_PROBE; + } else if (local.sum_ns_per_byte / static_cast<double>(local.count) * options.skip_read_factor > + remote.sum_ns_per_byte / static_cast<double>(remote.count)) { + // A real local probe lets an OS-cache hit or recovered disk replace old slow samples. + // For example, one in 32 otherwise bypassed reads still measures the local path. + decision.route = + bucket.decisions % options.probe_interval == 0 ? Route::LOCAL_PROBE : Route::REMOTE; + } + if (!decision.read_remote()) { + return decision; + } + int inflight = _remote_inflight.load(std::memory_order_relaxed); + while (inflight < options.max_remote_inflight) { + if (_remote_inflight.compare_exchange_weak(inflight, inflight + 1, + std::memory_order_relaxed)) { + decision._permit = &_remote_inflight; + return decision; + } + } + decision.route = Route::LOCAL_BUDGET; + return decision; +} + +void FileCacheIOAdaptor::record_local(uint64_t resource, size_t bytes, int64_t elapsed_ns, + int64_t now_ns, const Options& options) { + auto& bucket = _buckets[bucket_index(resource, bytes)]; + std::lock_guard lock(bucket.mutex); + bucket.local.expire(now_ns, options.sample_ttl_ns); + bucket.local.record(bytes, elapsed_ns, now_ns); +} + +void FileCacheIOAdaptor::record_remote(uint64_t resource, size_t bytes, int64_t elapsed_ns, + int64_t now_ns, const Options& options) { + auto& bucket = _buckets[bucket_index(resource, bytes)]; + std::lock_guard lock(bucket.mutex); + bucket.remote.expire(now_ns, options.sample_ttl_ns); + bucket.remote.record(bytes, elapsed_ns, now_ns); +} + +void FileCacheIOAdaptor::record_remote_failure(uint64_t resource, size_t bytes, int64_t now_ns, + const Options& options) { + auto& bucket = _buckets[bucket_index(resource, bytes)]; + std::lock_guard lock(bucket.mutex); + bucket.remote_failure_until_ns = now_ns + options.remote_cooldown_ns; +} +} // namespace doris::io diff --git a/be/src/io/cache/file_cache_io_adaptor.h b/be/src/io/cache/file_cache_io_adaptor.h new file mode 100644 index 00000000000..d63e43dc3e9 --- /dev/null +++ b/be/src/io/cache/file_cache_io_adaptor.h @@ -0,0 +1,117 @@ +// 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. + +#pragma once + +#include <array> +#include <atomic> +#include <cstddef> +#include <cstdint> +#include <mutex> +#include <utility> + +namespace doris::io { + +// A cache-local controller: bounded resource/size shards avoid per-file state and worker threads. +class FileCacheIOAdaptor { +public: + struct Options { + size_t min_samples = 50; + int64_t min_local_latency_ns = 5'000'000; + double skip_read_factor = 1.0; + uint64_t probe_interval = 32; + int64_t sample_ttl_ns = 30'000'000'000; + int64_t remote_cooldown_ns = 1'000'000'000; + int max_remote_inflight = 8; + }; + enum class Route { LOCAL, REMOTE, LOCAL_PROBE, REMOTE_PROBE, LOCAL_BUDGET }; + class Decision { + public: + Decision() = default; + ~Decision() { release(); } + Decision(const Decision&) = delete; + Decision& operator=(const Decision&) = delete; + Decision(Decision&& other) noexcept + : route(other.route), + local_latency_ns(other.local_latency_ns), + remote_latency_ns(other.remote_latency_ns), + _permit(std::exchange(other._permit, nullptr)) {} + Decision& operator=(Decision&& other) noexcept { + if (this != &other) { + release(); + route = other.route; + local_latency_ns = other.local_latency_ns; + remote_latency_ns = other.remote_latency_ns; + _permit = std::exchange(other._permit, nullptr); + } + return *this; + } + bool read_remote() const { return route == Route::REMOTE || route == Route::REMOTE_PROBE; } + Route route = Route::LOCAL; + int64_t local_latency_ns = 0; + int64_t remote_latency_ns = 0; + + private: + friend class FileCacheIOAdaptor; + void release() { + if (_permit != nullptr) { + _permit->fetch_sub(1, std::memory_order_relaxed); + _permit = nullptr; + } + } + std::atomic<int>* _permit = nullptr; + }; + + Decision select(uint64_t resource, size_t bytes, int64_t now_ns, const Options& options); + void record_local(uint64_t resource, size_t bytes, int64_t elapsed_ns, int64_t now_ns, + const Options& options); + void record_remote(uint64_t resource, size_t bytes, int64_t elapsed_ns, int64_t now_ns, + const Options& options); + void record_remote_failure(uint64_t resource, size_t bytes, int64_t now_ns, + const Options& options); + +private: + static constexpr size_t WINDOW_SIZE = 50; + static constexpr size_t RESOURCE_SHARDS = 16; + static constexpr size_t SIZE_BUCKETS = 8; + struct Sample { + int64_t latency_ns = 0; + double ns_per_byte = 0; + }; + struct Window { + std::array<Sample, WINDOW_SIZE> samples {}; + size_t count = 0; + size_t next = 0; + int64_t sum_ns = 0; + double sum_ns_per_byte = 0; + int64_t last_ns = 0; + void expire(int64_t now_ns, int64_t ttl_ns); + void record(size_t bytes, int64_t elapsed_ns, int64_t now_ns); + }; + struct Bucket { + std::mutex mutex; + Window local; + Window remote; + uint64_t decisions = 0; + int64_t remote_failure_until_ns = 0; + }; + static size_t bucket_index(uint64_t resource, size_t bytes); + std::array<Bucket, RESOURCE_SHARDS * SIZE_BUCKETS> _buckets; + std::atomic<int> _remote_inflight {0}; +}; + +} // namespace doris::io diff --git a/be/src/io/io_common.h b/be/src/io/io_common.h index 0109a345af3..13bfdd032eb 100644 --- a/be/src/io/io_common.h +++ b/be/src/io/io_common.h @@ -73,6 +73,18 @@ struct FileCacheStatistics { int64_t lock_wait_timer = 0; int64_t get_timer = 0; int64_t set_timer = 0; + int64_t io_adaptor_lookup_count = 0; + int64_t io_adaptor_lookup_time_ns = 0; + int64_t io_adaptor_local_reads = 0; + int64_t io_adaptor_local_time_ns = 0; + int64_t io_adaptor_bypass_count = 0; + int64_t io_adaptor_bypass_bytes = 0; + int64_t io_adaptor_remote_bytes = 0; + int64_t io_adaptor_remote_time_ns = 0; + int64_t io_adaptor_local_probes = 0; + int64_t io_adaptor_remote_probes = 0; + int64_t io_adaptor_budget_rejected = 0; + int64_t io_adaptor_fallback_local = 0; int64_t async_cache_write_submitted = 0; int64_t async_cache_write_rejected = 0; int64_t async_cache_write_buffer_alloc_fail = 0; @@ -145,6 +157,18 @@ struct FileCacheStatistics { lock_wait_timer += other.lock_wait_timer; get_timer += other.get_timer; set_timer += other.set_timer; + io_adaptor_lookup_count += other.io_adaptor_lookup_count; + io_adaptor_lookup_time_ns += other.io_adaptor_lookup_time_ns; + io_adaptor_local_reads += other.io_adaptor_local_reads; + io_adaptor_local_time_ns += other.io_adaptor_local_time_ns; + io_adaptor_bypass_count += other.io_adaptor_bypass_count; + io_adaptor_bypass_bytes += other.io_adaptor_bypass_bytes; + io_adaptor_remote_bytes += other.io_adaptor_remote_bytes; + io_adaptor_remote_time_ns += other.io_adaptor_remote_time_ns; + io_adaptor_local_probes += other.io_adaptor_local_probes; + io_adaptor_remote_probes += other.io_adaptor_remote_probes; + io_adaptor_budget_rejected += other.io_adaptor_budget_rejected; + io_adaptor_fallback_local += other.io_adaptor_fallback_local; async_cache_write_submitted += other.async_cache_write_submitted; async_cache_write_rejected += other.async_cache_write_rejected; async_cache_write_buffer_alloc_fail += other.async_cache_write_buffer_alloc_fail; diff --git a/be/test/io/cache/block_file_cache_profile_reporter_test.cpp b/be/test/io/cache/block_file_cache_profile_reporter_test.cpp index 4836bcb2638..834d0077d96 100644 --- a/be/test/io/cache/block_file_cache_profile_reporter_test.cpp +++ b/be/test/io/cache/block_file_cache_profile_reporter_test.cpp @@ -54,6 +54,18 @@ io::FileCacheStatistics make_file_cache_stats(int64_t multiplier) { stats.inverted_index_io_timer = multiplier * 28; stats.remote_only_on_miss_triggered = multiplier * 29; stats.remote_only_on_miss_threshold_bytes = multiplier * 30; + stats.io_adaptor_lookup_count = multiplier * 60; + stats.io_adaptor_lookup_time_ns = multiplier * 61; + stats.io_adaptor_local_reads = multiplier * 62; + stats.io_adaptor_local_time_ns = multiplier * 63; + stats.io_adaptor_bypass_count = multiplier * 64; + stats.io_adaptor_bypass_bytes = multiplier * 65; + stats.io_adaptor_remote_bytes = multiplier * 66; + stats.io_adaptor_remote_time_ns = multiplier * 67; + stats.io_adaptor_local_probes = multiplier * 68; + stats.io_adaptor_remote_probes = multiplier * 69; + stats.io_adaptor_budget_rejected = multiplier * 70; + stats.io_adaptor_fallback_local = multiplier * 71; stats.async_cache_write_submitted = multiplier * 45; stats.async_cache_write_rejected = multiplier * 46; stats.async_cache_write_buffer_alloc_fail = multiplier * 47; @@ -70,6 +82,18 @@ io::FileCacheStatistics make_file_cache_stats(int64_t multiplier) { void expect_file_cache_stats_eq(const io::FileCacheStatistics& actual, const io::FileCacheStatistics& expected) { + EXPECT_EQ(actual.io_adaptor_lookup_count, expected.io_adaptor_lookup_count); + EXPECT_EQ(actual.io_adaptor_lookup_time_ns, expected.io_adaptor_lookup_time_ns); + EXPECT_EQ(actual.io_adaptor_local_reads, expected.io_adaptor_local_reads); + EXPECT_EQ(actual.io_adaptor_local_time_ns, expected.io_adaptor_local_time_ns); + EXPECT_EQ(actual.io_adaptor_bypass_count, expected.io_adaptor_bypass_count); + EXPECT_EQ(actual.io_adaptor_bypass_bytes, expected.io_adaptor_bypass_bytes); + EXPECT_EQ(actual.io_adaptor_remote_bytes, expected.io_adaptor_remote_bytes); + EXPECT_EQ(actual.io_adaptor_remote_time_ns, expected.io_adaptor_remote_time_ns); + EXPECT_EQ(actual.io_adaptor_local_probes, expected.io_adaptor_local_probes); + EXPECT_EQ(actual.io_adaptor_remote_probes, expected.io_adaptor_remote_probes); + EXPECT_EQ(actual.io_adaptor_budget_rejected, expected.io_adaptor_budget_rejected); + EXPECT_EQ(actual.io_adaptor_fallback_local, expected.io_adaptor_fallback_local); EXPECT_EQ(actual.num_local_io_total, expected.num_local_io_total); EXPECT_EQ(actual.num_remote_io_total, expected.num_remote_io_total); EXPECT_EQ(actual.num_peer_io_total, expected.num_peer_io_total); @@ -165,6 +189,30 @@ TEST(FileCacheProfileReporterTest, ReporterAggregatesDeltaReportsToExactFinalTot io::diff_file_cache_statistics(after_second_report, after_first_report); reporter.update(&second_delta); + EXPECT_EQ(profile->get_counter("IOAdaptorLookupCount")->value(), + after_second_report.io_adaptor_lookup_count); + EXPECT_EQ(profile->get_counter("IOAdaptorLookupTime")->value(), + after_second_report.io_adaptor_lookup_time_ns); + EXPECT_EQ(profile->get_counter("IOAdaptorLocalReadCount")->value(), + after_second_report.io_adaptor_local_reads); + EXPECT_EQ(profile->get_counter("IOAdaptorLocalReadTime")->value(), + after_second_report.io_adaptor_local_time_ns); + EXPECT_EQ(profile->get_counter("IOAdaptorBypassCount")->value(), + after_second_report.io_adaptor_bypass_count); + EXPECT_EQ(profile->get_counter("IOAdaptorBypassBytes")->value(), + after_second_report.io_adaptor_bypass_bytes); + EXPECT_EQ(profile->get_counter("IOAdaptorRemoteReadBytes")->value(), + after_second_report.io_adaptor_remote_bytes); + EXPECT_EQ(profile->get_counter("IOAdaptorRemoteReadTime")->value(), + after_second_report.io_adaptor_remote_time_ns); + EXPECT_EQ(profile->get_counter("IOAdaptorLocalProbeCount")->value(), + after_second_report.io_adaptor_local_probes); + EXPECT_EQ(profile->get_counter("IOAdaptorRemoteProbeCount")->value(), + after_second_report.io_adaptor_remote_probes); + EXPECT_EQ(profile->get_counter("IOAdaptorBudgetRejectedCount")->value(), + after_second_report.io_adaptor_budget_rejected); + EXPECT_EQ(profile->get_counter("IOAdaptorFallbackLocalCount")->value(), + after_second_report.io_adaptor_fallback_local); EXPECT_EQ(profile->get_counter("BytesScannedFromCache")->value(), after_second_report.bytes_read_from_local); EXPECT_EQ(profile->get_counter("BytesScannedFromRemote")->value(), diff --git a/be/test/io/cache/cached_remote_file_reader_test.cpp b/be/test/io/cache/cached_remote_file_reader_test.cpp index 6fe9bac5d60..9dc7d802c8c 100644 --- a/be/test/io/cache/cached_remote_file_reader_test.cpp +++ b/be/test/io/cache/cached_remote_file_reader_test.cpp @@ -15,12 +15,17 @@ // specific language governing permissions and limitations // under the License. +#include <algorithm> +#include <chrono> #include <cstring> #include <future> +#include <iostream> +#include <thread> #include "block_file_cache_test_common.h" #include "cloud/config.h" #include "io/cache/async_cache_write_manager_metrics.h" +#include "io/cache/file_cache_io_adaptor.h" #include "io/cache/inflight_write_buffer_index.h" #include "util/time.h" @@ -57,6 +62,10 @@ public: bool closed() const override { return _delegate->closed(); } int64_t mtime() const override { return _delegate->mtime(); } + int64_t delay_us = 0; + bool fail_reads = false; + bool cancel_reads = false; + size_t short_read_size = 0; size_t read_count() const { return _read_count; } size_t last_offset() const { return _last_offset; } size_t last_size() const { return _last_size; } @@ -69,6 +78,18 @@ protected: _last_offset = offset; _last_size = result.size; _last_bypass_peer_read = io_ctx != nullptr && io_ctx->bypass_peer_read; + if (delay_us > 0) { + std::this_thread::sleep_for(std::chrono::microseconds(delay_us)); + } + if (cancel_reads) { + return Status::Cancelled("Injected remote cancellation"); + } + if (fail_reads) { + return Status::IOError("Injected remote read failure"); + } + if (short_read_size != 0) { + result = Slice(result.data, std::min(result.size, short_read_size)); + } return _delegate->read_at(offset, result, bytes_read, io_ctx); } @@ -1661,4 +1682,415 @@ TEST_F(BlockFileCacheTest, cache_write_mode_is_resolved_for_each_read_context) { EXPECT_EQ(sync_reader->_resolve_cache_write_mode(&context), CacheWriteMode::ASYNC_WRITE); } +// All-hit adaptation is shared by sync/async readers. Cold and mixed requests retain their +// original downloader/inflight/writeback semantics rather than pretending a cache hit is EMPTY. +class AdaptiveCachedRemoteFileReaderTest : public AsyncCachedRemoteFileReaderTest { +protected: + void SetUp() override { + AsyncCachedRemoteFileReaderTest::SetUp(); + old_enabled = config::enable_file_cache_io_adaptor; + config::enable_file_cache_io_adaptor = false; + } + void TearDown() override { + config::enable_file_cache_io_adaptor = old_enabled; + AsyncCachedRemoteFileReaderTest::TearDown(); + } + std::shared_ptr<CachedRemoteFileReader> external_reader(FileReaderSPtr remote) { + FileReaderOptions options; + options.cache_type = FileCachePolicy::FILE_BLOCK_CACHE; + return std::make_shared<CachedRemoteFileReader>(std::move(remote), options); + } + void seed_slow_disk(const std::shared_ptr<CachedRemoteFileReader>& reader, size_t bytes) { + FileCacheIOAdaptor::Options options; + for (int i = 0; i < 50; ++i) { + cache()->io_adaptor().record_local(reader->_io_adaptor_resource_key, bytes, 20'000'000, + MonotonicNanos(), options); + cache()->io_adaptor().record_remote(reader->_io_adaptor_resource_key, bytes, 1'000'000, + MonotonicNanos(), options); + } + } + void warm_adaptor_with_real_reads(const std::shared_ptr<CachedRemoteFileReader>& reader, + const std::shared_ptr<CountingFileReader>& remote, + size_t bytes) { + // Each benchmark group starts with an isolated controller and measured service times. + // Calibration is excluded from the timed steady-state phase in both configurations. + cache()->_io_adaptor = std::make_unique<FileCacheIOAdaptor>(); + FileCacheIOAdaptor::Options options; + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + std::string buffer(bytes, '\0'); + size_t count = 0; + for (int sample = 0; sample < 50; ++sample) { + const auto start = MonotonicNanos(); + ASSERT_TRUE(remote->read_at(0, Slice(buffer), &count, &context).ok()); + ASSERT_EQ(count, bytes); + cache()->io_adaptor().record_remote(reader->_io_adaptor_resource_key, bytes, + MonotonicNanos() - start, MonotonicNanos(), + options); + } + config::enable_file_cache_io_adaptor = true; + for (int sample = 0; sample < 50; ++sample) { + ASSERT_TRUE(reader->read_at(0, Slice(buffer), &count, &context).ok()); + ASSERT_EQ(count, bytes); + } + } + bool old_enabled = false; +}; + +TEST_F(AdaptiveCachedRemoteFileReaderTest, full_hit_bypass_preserves_cache_in_both_write_modes) { + create_cache("adaptive_full_hit"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + std::string expected(64 * 1024, '\0'); + size_t count = 0; + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + ASSERT_TRUE(reader->read_at(17, Slice(expected), &count, &context).ok()); + ASSERT_EQ(count, expected.size()); + seed_slow_disk(reader, expected.size()); + config::enable_file_cache_io_adaptor = true; + const auto cached_blocks = cache()->get_blocks_by_key(reader->_cache_hash).size(); + for (auto mode : {CacheWriteMode::SYNC_WRITE, CacheWriteMode::ASYNC_WRITE}) { + FileCacheStatistics stats; + context.file_cache_stats = &stats; + context.cache_write_mode_override = mode; + const auto previous_reads = remote->read_count(); + std::string actual(expected.size(), '\0'); + ASSERT_TRUE(reader->read_at(17, Slice(actual), &count, &context).ok()); + EXPECT_EQ(actual, expected); + EXPECT_EQ(remote->read_count(), previous_reads + 1); + EXPECT_EQ(remote->last_offset(), 17); + EXPECT_EQ(remote->last_size(), actual.size()); + EXPECT_EQ(stats.io_adaptor_bypass_count, 1); + EXPECT_EQ(stats.io_adaptor_bypass_bytes, actual.size()); + EXPECT_EQ(stats.io_adaptor_remote_bytes, actual.size()); + EXPECT_EQ(stats.bytes_read_from_remote, actual.size()); + EXPECT_EQ(stats.bytes_write_into_cache, 0); + EXPECT_EQ(stats.async_cache_write_submitted, 0); + EXPECT_EQ(cache()->get_blocks_by_key(reader->_cache_hash).size(), cached_blocks); + } +} + +TEST_F(AdaptiveCachedRemoteFileReaderTest, cross_block_and_eof_ranges_match_on_both_routes) { + create_cache("adaptive_read_ranges"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + for (size_t offset : {size_t(1_mb - 17), reader->size() - 31}) { + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + std::string expected(64 * 1024, '\0'); + size_t count = 0; + ASSERT_TRUE(reader->read_at(offset, Slice(expected), &count, &context).ok()); + expected.resize(count); + config::enable_file_cache_io_adaptor = true; + for (bool bypass : {false, true}) { + if (bypass) { + seed_slow_disk(reader, expected.size()); + } + const auto previous_reads = remote->read_count(); + FileCacheStatistics stats; + context.file_cache_stats = &stats; + std::string actual(64 * 1024, '\xff'); + ASSERT_TRUE(reader->read_at(offset, Slice(actual), &count, &context).ok()); + EXPECT_EQ(count, expected.size()); + EXPECT_EQ(actual.substr(0, count), expected); + EXPECT_EQ(actual.substr(count), std::string(actual.size() - count, '\xff')); + EXPECT_EQ(remote->read_count(), previous_reads + size_t(bypass)); + EXPECT_EQ(stats.io_adaptor_bypass_count, int64_t(bypass)); + EXPECT_EQ(stats.io_adaptor_local_reads, int64_t(!bypass)); + EXPECT_EQ(stats.bytes_write_into_cache, 0); + } + } +} + +TEST_F(AdaptiveCachedRemoteFileReaderTest, disabled_and_metadata_reads_preserve_local_path) { + create_cache("adaptive_disabled"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + std::string buffer(64 * 1024, '\0'); + size_t count = 0; + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + ASSERT_TRUE(reader->read_at(0, Slice(buffer), &count, &context).ok()); + seed_slow_disk(reader, buffer.size()); + const auto previous_reads = remote->read_count(); + ASSERT_TRUE(reader->read_at(0, Slice(buffer), &count, &context).ok()); + EXPECT_EQ(remote->read_count(), previous_reads); + config::enable_file_cache_io_adaptor = true; + context.is_index_data = true; + EXPECT_FALSE(reader->_io_adaptor_eligible(buffer.size(), &context)); + ASSERT_TRUE(reader->read_at(0, Slice(buffer), &count, &context).ok()); + EXPECT_EQ(remote->read_count(), previous_reads); + context.is_index_data = false; + context.is_warmup = true; + EXPECT_FALSE(reader->_io_adaptor_eligible(buffer.size(), &context)); + context.is_warmup = false; + context.is_dryrun = true; + EXPECT_FALSE(reader->_io_adaptor_eligible(buffer.size(), &context)); + context.is_dryrun = false; + context.should_stop = true; + EXPECT_FALSE(reader->_io_adaptor_eligible(buffer.size(), &context)); + context.should_stop = false; + EXPECT_FALSE(reader->_io_adaptor_eligible( + static_cast<size_t>(config::file_cache_io_adaptor_max_request_bytes) + 1, &context)); + EXPECT_FALSE(reader->_io_adaptor_eligible(0, &context)); +} + +TEST_F(AdaptiveCachedRemoteFileReaderTest, remote_failure_and_short_read_fall_back_to_valid_cache) { + create_cache("adaptive_remote_failure"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + std::string expected(64 * 1024, '\0'); + size_t count = 0; + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + ASSERT_TRUE(reader->read_at(7, Slice(expected), &count, &context).ok()); + seed_slow_disk(reader, expected.size()); + config::enable_file_cache_io_adaptor = true; + FileCacheStatistics stats; + context.file_cache_stats = &stats; + remote->short_read_size = 7; + std::string actual(expected.size(), '\xff'); + ASSERT_TRUE(reader->read_at(7, Slice(actual), &count, &context).ok()); + EXPECT_EQ(actual, expected); + EXPECT_EQ(count, expected.size()); + EXPECT_EQ(stats.io_adaptor_fallback_local, 1); + EXPECT_EQ(stats.io_adaptor_remote_bytes, 7); + EXPECT_EQ(stats.bytes_read_from_local, expected.size()); + EXPECT_EQ(stats.bytes_write_into_cache, 0); + // Failure cooldown suppresses a second speculative remote read. + remote->short_read_size = 0; + remote->fail_reads = true; + const auto previous_reads = remote->read_count(); + ASSERT_TRUE(reader->read_at(7, Slice(actual), &count, &context).ok()); + EXPECT_EQ(remote->read_count(), previous_reads); + EXPECT_EQ(actual, expected); + FileCacheIOAdaptor::Options expired_cooldown; + expired_cooldown.remote_cooldown_ns = 0; + cache()->io_adaptor().record_remote_failure(reader->_io_adaptor_resource_key, actual.size(), + MonotonicNanos(), expired_cooldown); + ASSERT_TRUE(reader->read_at(7, Slice(actual), &count, &context).ok()); + EXPECT_EQ(remote->read_count(), previous_reads + 1); + EXPECT_EQ(stats.io_adaptor_fallback_local, 2); + EXPECT_EQ(actual, expected); +} + +TEST_F(AdaptiveCachedRemoteFileReaderTest, cancellation_does_not_fall_back_or_rewrite_cache) { + create_cache("adaptive_cancelled"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + std::string buffer(64 * 1024, '\0'); + size_t count = 0; + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + ASSERT_TRUE(reader->read_at(0, Slice(buffer), &count, &context).ok()); + seed_slow_disk(reader, buffer.size()); + config::enable_file_cache_io_adaptor = true; + remote->cancel_reads = true; + const auto previous_reads = remote->read_count(); + auto status = reader->read_at(0, Slice(buffer), &count, &context); + EXPECT_TRUE(status.is<ErrorCode::CANCELLED>()); + EXPECT_EQ(remote->read_count(), previous_reads + 1); + // Cancellation is propagated, while a subsequent query can still use the intact cache. + remote->cancel_reads = false; + const auto reads_after_cancel = remote->read_count(); + ASSERT_TRUE(reader->read_at(0, Slice(buffer), &count, &context).ok()); + EXPECT_EQ(remote->read_count(), reads_after_cancel); + EXPECT_EQ(count, buffer.size()); +} + +TEST_F(AdaptiveCachedRemoteFileReaderTest, mixed_hit_miss_and_eof_keep_existing_async_semantics) { + create_cache("adaptive_mixed_range"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + std::string first(1_mb, '\0'); + size_t count = 0; + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + ASSERT_TRUE(reader->read_at(0, Slice(first), &count, &context).ok()); + seed_slow_disk(reader, 2_mb); + config::enable_file_cache_io_adaptor = true; + context.cache_write_mode_override = CacheWriteMode::ASYNC_WRITE; + FileCacheStatistics stats; + context.file_cache_stats = &stats; + std::string actual(2_mb, '\0'); + ASSERT_TRUE(reader->read_at(0, Slice(actual), &count, &context).ok()); + EXPECT_EQ(count, actual.size()); + EXPECT_EQ(actual.substr(0, first.size()), first); + EXPECT_EQ(stats.io_adaptor_bypass_count, 0); + EXPECT_GT(stats.async_cache_write_submitted, 0); + wait_for_async_writes(); + ASSERT_TRUE(reader->read_at(reader->size(), Slice(actual), &count, &context).ok()); + EXPECT_EQ(count, 0); +} + +TEST_F(AdaptiveCachedRemoteFileReaderTest, missing_cache_file_retries_the_complete_range) { + create_cache("adaptive_missing_file"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + std::string expected(64 * 1024, '\0'); + size_t count = 0; + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + ASSERT_TRUE(reader->read_at(0, Slice(expected), &count, &context).ok()); + auto blocks = cache()->get_blocks_by_key(reader->_cache_hash); + ASSERT_FALSE(blocks.empty()); + std::error_code error; + ASSERT_TRUE(fs::remove(blocks.begin()->second->get_cache_file(), error)); + ASSERT_FALSE(error); + blocks.clear(); + config::enable_file_cache_io_adaptor = true; + FileCacheStatistics stats; + context.file_cache_stats = &stats; + const auto previous_reads = remote->read_count(); + std::string actual(expected.size(), '\xff'); + ASSERT_TRUE(reader->read_at(0, Slice(actual), &count, &context).ok()); + EXPECT_EQ(actual, expected); + EXPECT_EQ(count, actual.size()); + EXPECT_EQ(remote->read_count(), previous_reads + 1); + EXPECT_EQ(stats.io_adaptor_local_reads, 1); + EXPECT_GT(stats.io_adaptor_local_time_ns, 0); + EXPECT_EQ(stats.io_adaptor_bypass_count, 0); +} + +TEST_F(AdaptiveCachedRemoteFileReaderTest, inflight_payload_still_avoids_remote_reads) { + create_cache("adaptive_inflight"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + auto* index = cache()->inflight_write_buffer_index(); + AsyncCacheWriteBufferPtr payload; + ASSERT_TRUE(cache()->async_write_manager()->allocate_tracked_buffer(1_mb, &payload).ok()); + const std::string expected(1_mb, '0'); + memcpy(payload->data(), expected.data(), expected.size()); + auto entry = std::make_shared<InflightWriteBufferEntry>(payload, 0, expected.size(), + MonotonicMicros()); + ASSERT_EQ(index->insert_if_absent(reader->_cache_hash, 0, entry), nullptr); + Defer remove_entry { + [&] { static_cast<void>(index->remove_if(reader->_cache_hash, 0, entry)); }}; + seed_slow_disk(reader, expected.size()); + config::enable_file_cache_io_adaptor = true; + FileCacheStatistics stats; + IOContext context; + context.file_cache_stats = &stats; + context.cache_write_mode_override = CacheWriteMode::ASYNC_WRITE; + std::string actual(expected.size(), '\0'); + size_t count = 0; + ASSERT_TRUE(reader->read_at(0, Slice(actual), &count, &context).ok()); + EXPECT_EQ(actual, expected); + EXPECT_EQ(count, actual.size()); + EXPECT_EQ(remote->read_count(), 0); + EXPECT_EQ(stats.io_adaptor_bypass_count, 0); + EXPECT_GT(stats.inflight_write_buffer_index_hit, 0); + EXPECT_EQ(stats.async_cache_write_submitted, 0); +} + +// Opt-in end-to-end cache-path benchmark. Injected service times model a congested cache disk; +// the measured wall times are a controlled local experiment, not a production query forecast. +TEST_F(AdaptiveCachedRemoteFileReaderTest, DISABLED_latency_controlled_benchmark) { + create_cache("adaptive_latency_benchmark"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + std::string expected(64 * 1024, '\0'); + size_t count = 0; + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + ASSERT_TRUE(reader->read_at(0, Slice(expected), &count, &context).ok()); + remote->delay_us = 1000; + auto* sync_point = SyncPoint::get_instance(); + SyncPoint::CallbackGuard guard; + sync_point->set_call_back( + "FileBlock::adaptor_benchmark_read", + [](auto&&) { std::this_thread::sleep_for(std::chrono::microseconds(8000)); }, &guard); + sync_point->enable_processing(); + Defer cleanup {[&] { sync_point->disable_processing(); }}; + std::vector<int64_t> off_times; + std::vector<int64_t> on_times; + for (int round = 0; round < 5; ++round) { + for (int phase = 0; phase < 2; ++phase) { + const bool enabled = (round + phase) % 2 == 1; + ASSERT_NO_FATAL_FAILURE(warm_adaptor_with_real_reads(reader, remote, expected.size())); + config::enable_file_cache_io_adaptor = enabled; + FileCacheStatistics stats; + context.file_cache_stats = &stats; + std::string actual(expected.size(), '\0'); + const auto start = MonotonicNanos(); + for (int request = 0; request < 200; ++request) { + ASSERT_TRUE(reader->read_at(0, Slice(actual), &count, &context).ok()); + ASSERT_EQ(actual, expected); + ASSERT_EQ(count, expected.size()); + } + const auto elapsed = MonotonicNanos() - start; + (enabled ? on_times : off_times).push_back(elapsed); + std::cout << "IO_ADAPTOR_BENCH round=" << round << " enabled=" << enabled + << " requests=200 request_bytes=" << expected.size() + << " local_service_us=8000 remote_service_us=1000 wall_ns=" << elapsed + << " bypass_count=" << stats.io_adaptor_bypass_count + << " bypass_bytes=" << stats.io_adaptor_bypass_bytes + << " local_probe_count=" << stats.io_adaptor_local_probes << std::endl; + EXPECT_EQ(stats.bytes_write_into_cache, 0); + if (enabled) { + EXPECT_GT(stats.io_adaptor_bypass_count, 180); + EXPECT_GT(stats.io_adaptor_local_probes, 0); + } else { + EXPECT_EQ(stats.io_adaptor_bypass_count, 0); + } + } + } + std::sort(off_times.begin(), off_times.end()); + std::sort(on_times.begin(), on_times.end()); + std::cout << "IO_ADAPTOR_BENCH_MEDIAN off_ns=" << off_times[2] << " on_ns=" << on_times[2] + << std::endl; +} + +TEST_F(AdaptiveCachedRemoteFileReaderTest, DISABLED_fast_local_latency_controlled_benchmark) { + create_cache("adaptive_fast_local_benchmark"); + auto remote = std::make_shared<CountingFileReader>(open_remote_file()); + auto reader = external_reader(remote); + std::string expected(64 * 1024, '\0'); + size_t count = 0; + IOContext context; + context.cache_write_mode_override = CacheWriteMode::SYNC_WRITE; + ASSERT_TRUE(reader->read_at(0, Slice(expected), &count, &context).ok()); + remote->delay_us = 8000; + auto* sync_point = SyncPoint::get_instance(); + SyncPoint::CallbackGuard guard; + sync_point->set_call_back( + "FileBlock::adaptor_benchmark_read", + [](auto&&) { std::this_thread::sleep_for(std::chrono::microseconds(1000)); }, &guard); + sync_point->enable_processing(); + Defer cleanup {[&] { sync_point->disable_processing(); }}; + std::vector<int64_t> off_times; + std::vector<int64_t> on_times; + for (int round = 0; round < 5; ++round) { + for (int phase = 0; phase < 2; ++phase) { + const bool enabled = (round + phase) % 2 == 1; + ASSERT_NO_FATAL_FAILURE(warm_adaptor_with_real_reads(reader, remote, expected.size())); + config::enable_file_cache_io_adaptor = enabled; + FileCacheStatistics stats; + context.file_cache_stats = &stats; + std::string actual(expected.size(), '\0'); + const auto previous_reads = remote->read_count(); + const auto start = MonotonicNanos(); + for (int request = 0; request < 200; ++request) { + ASSERT_TRUE(reader->read_at(0, Slice(actual), &count, &context).ok()); + ASSERT_EQ(actual, expected); + ASSERT_EQ(count, expected.size()); + } + const auto elapsed = MonotonicNanos() - start; + (enabled ? on_times : off_times).push_back(elapsed); + std::cout << "IO_ADAPTOR_FAST_LOCAL_BENCH round=" << round << " enabled=" << enabled + << " requests=200 request_bytes=" << expected.size() + << " local_service_us=1000 remote_service_us=8000 wall_ns=" << elapsed + << " bypass_count=" << stats.io_adaptor_bypass_count << std::endl; + EXPECT_EQ(remote->read_count(), previous_reads); + EXPECT_EQ(stats.io_adaptor_bypass_count, 0); + EXPECT_EQ(stats.bytes_write_into_cache, 0); + } + } + std::sort(off_times.begin(), off_times.end()); + std::sort(on_times.begin(), on_times.end()); + std::cout << "IO_ADAPTOR_FAST_LOCAL_BENCH_MEDIAN off_ns=" << off_times[2] + << " on_ns=" << on_times[2] << std::endl; +} + } // namespace doris::io diff --git a/be/test/io/cache/file_cache_io_adaptor_test.cpp b/be/test/io/cache/file_cache_io_adaptor_test.cpp new file mode 100644 index 00000000000..b463aa61626 --- /dev/null +++ b/be/test/io/cache/file_cache_io_adaptor_test.cpp @@ -0,0 +1,182 @@ +// 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/file_cache_io_adaptor.h" + +#include <gtest/gtest.h> + +#include <thread> +#include <vector> + +namespace doris::io { + +class FileCacheIOAdaptorTest : public testing::Test { +protected: + void seed(size_t count, int64_t local_ns = 20'000'000, int64_t remote_ns = 1'000'000) { + for (size_t i = 0; i < count; ++i) { + adaptor.record_local(0, bytes, local_ns, now, options); + adaptor.record_remote(0, bytes, remote_ns, now, options); + } + } + FileCacheIOAdaptor adaptor; + FileCacheIOAdaptor::Options options; + static constexpr size_t bytes = 1 << 20; + static constexpr int64_t now = 1'000'000'000; +}; + +TEST_F(FileCacheIOAdaptorTest, BypassesSlowDiskAfterEnoughSamples) { + seed(49); + EXPECT_FALSE(adaptor.select(0, bytes, now, options).read_remote()); + seed(1); + EXPECT_TRUE(adaptor.select(0, bytes, now, options).read_remote()); +} + +TEST_F(FileCacheIOAdaptorTest, KeepsFastLocalReads) { + seed(50, 4'000'000, 1'000'000); + EXPECT_FALSE(adaptor.select(0, bytes, now, options).read_remote()); +} + +TEST_F(FileCacheIOAdaptorTest, KeepsLocalWhenRemoteIsSlower) { + seed(50, 8'000'000, 20'000'000); + EXPECT_FALSE(adaptor.select(0, bytes, now, options).read_remote()); +} + +TEST_F(FileCacheIOAdaptorTest, ThresholdAndFractionalFactor) { + seed(50, 8'000'000, 10'000'000); + EXPECT_FALSE(adaptor.select(0, bytes, now, options).read_remote()); + options.skip_read_factor = 1.5; + EXPECT_TRUE(adaptor.select(0, bytes, now, options).read_remote()); +} + +TEST_F(FileCacheIOAdaptorTest, EqualLatencyPrefersLocal) { + seed(50, 8'000'000, 8'000'000); + EXPECT_FALSE(adaptor.select(0, bytes, now, options).read_remote()); +} + +TEST_F(FileCacheIOAdaptorTest, ExactMinimumLatencyCanBypass) { + seed(50, options.min_local_latency_ns, 1'000'000); + EXPECT_TRUE(adaptor.select(0, bytes, now, options).read_remote()); +} + +TEST_F(FileCacheIOAdaptorTest, SeparatesResourcesAndReadSizes) { + seed(50); + EXPECT_FALSE(adaptor.select(1, bytes, now, options).read_remote()); + EXPECT_FALSE(adaptor.select(0, 4096, now, options).read_remote()); + EXPECT_TRUE(adaptor.select(0, bytes, now, options).read_remote()); +} + +TEST_F(FileCacheIOAdaptorTest, ComparesPerByteCostsWithinOneSizeBucket) { + constexpr size_t larger = 64 * 1024; + constexpr size_t smaller = 32 * 1024; + for (int i = 0; i < 50; ++i) { + adaptor.record_local(0, larger, 8'000'000, now, options); + adaptor.record_remote(0, smaller, 6'000'000, now, options); + } + // A smaller remote request takes less wall time but costs more per byte. + EXPECT_FALSE(adaptor.select(0, larger, now, options).read_remote()); + for (int i = 0; i < 50; ++i) { + adaptor.record_remote(0, larger, 6'000'000, now, options); + } + EXPECT_TRUE(adaptor.select(0, larger, now, options).read_remote()); +} + +TEST_F(FileCacheIOAdaptorTest, SlidingWindowReplacesSlowSamples) { + seed(50); + EXPECT_TRUE(adaptor.select(0, bytes, now, options).read_remote()); + seed(50, 100'000, 1'000'000); + EXPECT_FALSE(adaptor.select(0, bytes, now, options).read_remote()); +} + +TEST_F(FileCacheIOAdaptorTest, ExpiredSamplesDoNotDriveRouting) { + seed(50); + EXPECT_FALSE(adaptor.select(0, bytes, now + options.sample_ttl_ns, options).read_remote()); + adaptor.record_local(0, bytes, 20'000'000, now + options.sample_ttl_ns, options); + EXPECT_FALSE(adaptor.select(0, bytes, now + options.sample_ttl_ns, options).read_remote()); +} + +TEST_F(FileCacheIOAdaptorTest, WarmCacheBootstrapsWithBoundedRemoteProbes) { + for (size_t i = 0; i < 50; ++i) { + adaptor.record_local(0, bytes, 20'000'000, now, options); + } + for (size_t i = 1; i < options.probe_interval; ++i) { + EXPECT_FALSE(adaptor.select(0, bytes, now, options).read_remote()); + } + auto decision = adaptor.select(0, bytes, now, options); + EXPECT_EQ(decision.route, FileCacheIOAdaptor::Route::REMOTE_PROBE); + EXPECT_TRUE(decision.read_remote()); +} + +TEST_F(FileCacheIOAdaptorTest, PeriodicallyProbesLocalForRecovery) { + seed(50); + for (size_t i = 1; i < options.probe_interval; ++i) { + EXPECT_TRUE(adaptor.select(0, bytes, now, options).read_remote()); + } + EXPECT_EQ(adaptor.select(0, bytes, now, options).route, FileCacheIOAdaptor::Route::LOCAL_PROBE); +} + +TEST_F(FileCacheIOAdaptorTest, RemoteFailureTemporarilyKeepsLocal) { + seed(50); + adaptor.record_remote_failure(0, bytes, now, options); + EXPECT_FALSE(adaptor.select(0, bytes, now, options).read_remote()); + EXPECT_TRUE(adaptor.select(0, bytes, now + options.remote_cooldown_ns, options).read_remote()); +} + +TEST_F(FileCacheIOAdaptorTest, BudgetAndMoveReleaseExactlyOnce) { + seed(50); + options.max_remote_inflight = 1; + { + auto first = adaptor.select(0, bytes, now, options); + EXPECT_TRUE(first.read_remote()); + auto moved = std::move(first); + EXPECT_TRUE(moved.read_remote()); + EXPECT_EQ(adaptor.select(0, bytes, now, options).route, + FileCacheIOAdaptor::Route::LOCAL_BUDGET); + } + EXPECT_TRUE(adaptor.select(0, bytes, now, options).read_remote()); +} + +TEST_F(FileCacheIOAdaptorTest, ConcurrentBudgetIsBounded) { + seed(50); + options.max_remote_inflight = 4; + std::atomic<int> active {0}; + std::atomic<int> peak {0}; + std::vector<std::thread> threads; + for (int thread = 0; thread < 16; ++thread) { + threads.emplace_back([&] { + for (int i = 0; i < 2000; ++i) { + auto decision = adaptor.select(0, bytes, now, options); + if (decision.read_remote()) { + const auto current = active.fetch_add(1) + 1; + auto previous = peak.load(); + while (previous < current && !peak.compare_exchange_weak(previous, current)) { + } + std::this_thread::yield(); + active.fetch_sub(1); + } + } + }); + } + for (auto& thread : threads) { + thread.join(); + } + EXPECT_GT(peak.load(), 0); + EXPECT_LE(peak.load(), options.max_remote_inflight); + EXPECT_EQ(active.load(), 0); + EXPECT_TRUE(adaptor.select(0, bytes, now, options).read_remote()); +} + +} // namespace doris::io --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
