Gabriel39 commented on code in PR #69779: URL: https://github.com/apache/doris/pull/69779#discussion_r4490368793
########## be/src/io/cache/file_cache_io_adaptor.cpp: ########## @@ -0,0 +1,134 @@ +// 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 <algorithm> +#include <limits> + +#include "cpp/sync_point.h" +#include "util/time.h" + +namespace doris::io { + +size_t FileCacheIOAdaptor::bucket_index(size_t bytes) { + return static_cast<size_t>(std::ranges::lower_bound(SIZE_LIMITS, bytes) - SIZE_LIMITS.begin()); +} + +void FileCacheIOAdaptor::Window::record(int64_t latency_us) { + // Failed/invalid timings must not train the controller; a sub-microsecond read may be zero. + if (latency_us < 0) { + return; + } + if (count == WINDOW_SIZE) { + sum -= samples[next]; + } else { + ++count; + } + samples[next] = latency_us; + sum += latency_us; + next = (next + 1) % WINDOW_SIZE; +} + +int64_t FileCacheIOAdaptor::Window::average() const { + return count == 0 ? 0 : static_cast<int64_t>(sum / count); +} + +FileCacheIOAdaptor::Decision FileCacheIOAdaptor::select(size_t bytes, double skip_read_factor) { + auto now_us = MonotonicMicros(); + TEST_SYNC_POINT_CALLBACK("FileCacheIOAdaptor::select:now_us", &now_us); + return _select(bytes, skip_read_factor, now_us); +} + +FileCacheIOAdaptor::Decision FileCacheIOAdaptor::_select(size_t bytes, double skip_read_factor, + int64_t now_us) { + std::lock_guard lock(_mutex); + const auto index = bucket_index(bytes); + for (auto& candidate : _buckets) { + // Only seen buckets age; saturation prevents an idle bucket from wrapping into activity. + if (candidate.probe_wait_requests != 0 && + candidate.idle_requests != std::numeric_limits<size_t>::max()) { + ++candidate.idle_requests; + } + } + auto& bucket = _buckets[index]; + // With 64 small hits then one large hit, a fixed 64-hit expiry steals the large bucket's + // turn just before it returns. Retain its observed gap until it gets a probe; a short + // burst must not erase the longer wait either. + bucket.probe_wait_requests = + std::max({bucket.probe_wait_requests, bucket.idle_requests, PROBE_REQUEST_INTERVAL}); + bucket.idle_requests = 0; + Decision decision; + decision.local_latency_us = bucket.local.average(); + decision.remote_latency_us = bucket.remote.average(); + // Compare actual service times only within a size bucket: a 4 KiB / 100 us local read + // must not become a 12.8 ms full-block estimate and lose to an unrelated 1 MiB GET. + if (bucket.local.count == WINDOW_SIZE && bucket.remote.count == WINDOW_SIZE && + decision.local_latency_us >= MIN_LOCAL_LATENCY_US) { + decision.bypass = static_cast<double>(decision.local_latency_us) * skip_read_factor > + static_cast<double>(decision.remote_latency_us); + } + _requests_since_probe = std::min(_requests_since_probe + 1, PROBE_REQUEST_INTERVAL); + // History cannot reveal an unobserved recovery (e.g. local 20 ms -> 100 us). Route a real + // request through the other path, at most once per second and once per 64 eligible hits. + // Keep one probe in flight across the entire cache path, including slow local disk reads. + if (bucket.local.count == WINDOW_SIZE && !_probe_inflight && now_us >= _next_probe_us && + _requests_since_probe == PROBE_REQUEST_INTERVAL) { Review Comment: Fixed in 66a60cb0e031971a04692a2182cebfe1ee243fdf and synchronized to the branch-4.2 PR #69768 in fa7ed96761a2544a73252014cca4ff25bda657b9. Probe admission now requires a full local window with average latency at least MIN_LOCAL_LATENCY_US. Bucket rotation applies the same floor, so fast buckets neither send healthy cache hits to remote storage nor reserve a slow bucket's probe turn. Local recovery probes remain available while the measured local window is above the floor. Added three regression tests covering 1,024 reads at 32 requests/s, fast/slow bucket competition across eight probe intervals, and the exact 5 ms boundary. All three failed before the fix on both branches. The fast-local benchmark now requires zero remote reads, bypasses and probes. Failure, short-read and cancellation tests still verify probe quota consumption and permit release. All latest focused ASan results pass: 73 tests on master and 63 on branch-4.2, including both explicitly enabled benchmarks and the final cleanup-test reruns. clang-format 16 and changed-code clang-tidy passed. Both PRs have been queued for buildall and review again. -- 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]
