github-actions[bot] commented on code in PR #65658:
URL: https://github.com/apache/doris/pull/65658#discussion_r3709208902
##########
be/src/io/cache/block_file_cache.cpp:
##########
@@ -1370,6 +1503,8 @@ void BlockFileCache::try_evict_in_advance(size_t size,
std::lock_guard<std::mute
// remove specific cache synchronously, for critical operations
// if in use, cache meta will be deleted after use and the block file is then
deleted asynchronously
void BlockFileCache::remove_if_cached(const UInt128Wrapper& file_key) {
+ DORIS_CHECK(_async_write_service != nullptr);
+ _async_write_service->invalidate_pending_writes();
Review Comment:
[P1] Keep per-file invalidation scoped to this key. Both per-file removal
APIs advance the service-wide disk epoch, while workers discard every accepted
task carrying the old epoch regardless of its hash. Ordinary rowset/cache
recycling invokes this path once per segment and index, and the new test even
expects `queued_hash` to remain a gap after only `active_hash` is removed. A
recycle burst can therefore continuously cancel unrelated cache fills and
effectively disable async caching on that disk. Please use a per-key generation
for these APIs and reserve the global epoch for whole-cache invalidation.
##########
be/src/io/cache/async_cache_write_service.cpp:
##########
@@ -0,0 +1,731 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "io/cache/async_cache_write_service.h"
+
+#include <algorithm>
+#include <exception>
+#include <limits>
+#include <optional>
+#include <thread>
+#include <type_traits>
+#include <utility>
+
+#include "common/logging.h"
+#include "core/allocator.h"
+#include "cpp/sync_point.h"
+#include "io/cache/block_file_cache.h"
+#include "runtime/thread_context.h"
+#include "util/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
+
+Status resolve_async_file_cache_write_max_pending_bytes_per_disk(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 = 512LL * 1024 * 1024;
+ *resolved_bytes = static_cast<size_t>(std::max(kMinimumAutoPendingBytes,
be_mem_limit / 100));
+ return Status::OK();
+}
+
+class AsyncCacheWriteService::Worker : public
std::enable_shared_from_this<Worker> {
+public:
+ explicit Worker(AsyncCacheWriteService& service) : _service(service) {}
+
+ Status start() {
+ auto self = shared_from_this();
+ return _service._worker_pool->submit_func([self = std::move(self)]() {
self->_run(); });
+ }
+
+ void request_stop() { _stop_requested.store(true,
std::memory_order_release); }
+
+ void wait_until_stopped() { _stopped.wait(); }
+
+private:
+ void _run() {
+ _service._running_worker_count.fetch_add(1, std::memory_order_relaxed);
+ Defer mark_finished {[this]() {
+ const size_t old_running =
+ _service._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 (_service._try_take_task(&task)) {
+ _service._process_task(std::move(task));
+ continue;
+ }
+
+ if (_service._shutdown_requested.load(std::memory_order_acquire) &&
+ _service._pending_count.load(std::memory_order_acquire) == 0) {
+ return;
+ }
+ std::unique_lock lock(_service._queue_mutex);
+ _service._queue_cv.wait(lock, [this]() {
+ const bool shutdown_requested =
+
_service._shutdown_requested.load(std::memory_order_acquire);
+ return !_service._queue.empty() ||
+ _stop_requested.load(std::memory_order_acquire) ||
+ (shutdown_requested &&
+
_service._pending_count.load(std::memory_order_relaxed) == 0);
+ });
+ }
+ }
+
+ AsyncCacheWriteService& _service;
+ 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);
+}
+
+AsyncCacheWriteService::AsyncCacheWriteService(BlockFileCache* cache,
+ AsyncCacheWriteServiceOptions
options)
+ : _cache(cache),
+ _options(std::make_shared<const
AsyncCacheWriteServiceOptions>(options)),
+ _configured_worker_count(options.worker_count) {
+ DORIS_CHECK(_cache != nullptr);
+ DORIS_CHECK(options.worker_count > 0);
+ DORIS_CHECK(options.max_pending_bytes > 0);
+
+ const char* prefix = _cache->get_base_path().c_str();
+ _mem_tracker = MemTrackerLimiter::create_shared(
+ MemTrackerLimiter::Type::CACHE,
+ fmt::format("AsyncFileCacheWrite:{}", _cache->get_base_path()));
+ _pending_count_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_pending_count",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->pending_count();
+ },
+ this);
+ _pending_bytes_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_pending_bytes",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->pending_bytes();
+ },
+ this);
+ _queued_count_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_queue_size",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->queued_count();
+ },
+ this);
+ _queued_bytes_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_queued_bytes",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->queued_bytes();
+ },
+ this);
+ _active_task_count_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_active_tasks",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->active_task_count();
+ },
+ this);
+ _active_bytes_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_active_bytes",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->active_bytes();
+ },
+ this);
+ _running_worker_count_metric =
std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_running_workers",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->running_worker_count();
+ },
+ this);
+ _configured_worker_count_metric =
std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_configured_workers",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->_configured_worker_count.load(
+ std::memory_order_relaxed);
+ },
+ this);
+ _max_pending_bytes_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_max_pending_bytes",
+ [](void* service) {
+ return static_cast<AsyncCacheWriteService*>(service)
+ ->_options.load(std::memory_order_acquire)
+ ->max_pending_bytes;
+ },
+ this);
+ _active_get_or_set_count_metric =
std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_active_get_or_set",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->_active_get_or_set_count.load(
+ std::memory_order_relaxed);
+ },
+ this);
+ _active_append_count_metric =
std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_active_append",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->_active_append_count.load(
+ std::memory_order_relaxed);
+ },
+ this);
+ _active_finalize_count_metric =
std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_active_finalize",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->_active_finalize_count.load(
+ std::memory_order_relaxed);
+ },
+ this);
+ _buffer_memory_metric = std::make_shared<bvar::PassiveStatus<int64_t>>(
+ prefix, "async_cache_write_buffer_memory_bytes",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->buffer_memory_bytes();
+ },
+ this);
+ _submitted_metric =
+ std::make_shared<bvar::Adder<uint64_t>>(prefix,
"async_cache_write_submitted_total");
+ _submitted_bytes_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_submitted_bytes_total");
+ _finished_metric =
+ std::make_shared<bvar::Adder<uint64_t>>(prefix,
"async_cache_write_finished_total");
+ _finished_bytes_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_finished_bytes_total");
+ _worker_finished_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_worker_finished_total");
+ _worker_finished_bytes_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_worker_finished_bytes_total");
+ _evicted_oldest_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_evicted_oldest_total");
+ _evicted_oldest_bytes_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_evicted_oldest_bytes_total");
+ _evicted_oldest_age_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_evicted_oldest_age_us");
+ _rejected_metric =
+ std::make_shared<bvar::Adder<uint64_t>>(prefix,
"async_cache_write_rejected_total");
+ _reject_not_running_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_reject_not_running_total");
+ _reject_backpressure_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_reject_backpressure_total");
+ _buffer_alloc_fail_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_buffer_alloc_fail_total");
+ _submit_latency_metric =
+ std::make_shared<bvar::LatencyRecorder>(prefix,
"async_cache_write_submit_latency_us");
+ _buffer_alloc_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_buffer_alloc_latency_us");
+ _queue_wait_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_queue_wait_latency_us");
+ _queue_lock_wait_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_queue_lock_wait_latency_us");
+ _queue_lock_hold_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_queue_lock_hold_latency_us");
+ _worker_task_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_worker_task_latency_us");
+ _get_or_set_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_get_or_set_latency_us");
+ _append_latency_metric =
+ std::make_shared<bvar::LatencyRecorder>(prefix,
"async_cache_write_append_latency_us");
+ _finalize_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_finalize_latency_us");
+ _skip_downloaded_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_skip_downloaded_total");
+ _skip_downloading_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_skip_downloading_total");
+ _skip_partial_overlap_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_skip_partial_overlap_total");
+ _drop_stale_epoch_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_drop_stale_epoch_total");
+ _skip_deleting_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_skip_deleting_total");
+ _append_fail_metric =
+ std::make_shared<bvar::Adder<uint64_t>>(prefix,
"async_cache_write_append_fail_total");
+ _finalize_fail_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_finalize_fail_total");
+ _persisted_blocks_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_persisted_blocks_total");
+ _persisted_bytes_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_persisted_bytes_total");
+}
+
+AsyncCacheWriteService::~AsyncCacheWriteService() {
+ shutdown();
+}
+
+Status AsyncCacheWriteService::start() {
+ std::lock_guard resize_lock(_resize_mutex);
+ if (_shutdown_requested.load(std::memory_order_acquire) ||
+ !_accepting.load(std::memory_order_acquire)) {
+ return Status::InternalError("async file cache write service is
shutting down");
+ }
+ if (_started.load(std::memory_order_acquire)) {
+ return Status::OK();
+ }
+
+ const size_t worker_count =
_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 AsyncCacheWriteService::try_submit(AsyncCacheWriteTask task) {
+ DORIS_CHECK(task.buffer != nullptr);
+ DORIS_CHECK(task.write_size > 0);
+ DORIS_CHECK(task.write_size <= task.buffer->size());
+ DORIS_CHECK(task.write_size <= std::numeric_limits<size_t>::max() -
task.file_offset);
+ const int64_t submit_start_us = MonotonicMicros();
+ Defer record_submit_latency {
+ [&]() { *_submit_latency_metric << (MonotonicMicros() -
submit_start_us); }};
+ _active_submitters.fetch_add(1, std::memory_order_acq_rel);
+ Defer submitter_done {[&]() { _active_submitters.fetch_sub(1,
std::memory_order_acq_rel); }};
+
TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::try_submit:after_register",
&task);
+ if (!_started.load(std::memory_order_acquire) ||
!_accepting.load(std::memory_order_acquire)) {
+ *_rejected_metric << 1;
+ *_reject_not_running_metric << 1;
+ return false;
+ }
+
+ const size_t task_buffer_bytes = task.buffer->size();
+ std::optional<AsyncCacheWriteTask> victim;
+ {
+ TimedQueueLock lock(_queue_mutex, *_queue_lock_wait_latency_metric,
+ *_queue_lock_hold_latency_metric);
+ const auto options = _options.load(std::memory_order_acquire);
+ const size_t max_pending_bytes = options->max_pending_bytes;
+ const size_t pending_bytes =
_pending_bytes.load(std::memory_order_relaxed);
+ if (_task_buffer_size == 0) {
+ _task_buffer_size = task_buffer_bytes;
+ }
+ DORIS_CHECK(task_buffer_bytes == _task_buffer_size);
+
+ if (task_buffer_bytes > max_pending_bytes) {
+ *_rejected_metric << 1;
+ *_reject_backpressure_metric << 1;
+ return false;
+ }
+
+ const bool has_capacity = pending_bytes <= max_pending_bytes -
task_buffer_bytes;
+ if (!has_capacity && _queue.empty()) {
+ *_rejected_metric << 1;
+ *_reject_backpressure_metric << 1;
+ return false;
+ }
+
+ _queue.push_back(std::move(task));
+ if (has_capacity) {
+ _queued_bytes.fetch_add(task_buffer_bytes,
std::memory_order_relaxed);
+ _pending_count.fetch_add(1, std::memory_order_relaxed);
+ _pending_bytes.fetch_add(task_buffer_bytes,
std::memory_order_relaxed);
+ } else {
+ victim.emplace(std::move(_queue.front()));
+ _queue.pop_front();
+ }
+ }
+
+ *_submitted_metric << 1;
+ *_submitted_bytes_metric << task_buffer_bytes;
+ _queue_cv.notify_one();
+ if (victim) {
+ _finalize_task(std::move(*victim),
TaskFinalizationReason::EVICTED_OLDEST);
+ }
+ return true;
+}
+
+Status AsyncCacheWriteService::allocate_tracked_buffer(size_t size,
+
AsyncCacheWriteBufferPtr* buffer) {
+ DORIS_CHECK(buffer != nullptr);
+ DORIS_CHECK(size > 0);
+ const int64_t allocation_start_us = MonotonicMicros();
+ Defer record_allocation_latency {
+ [&]() { *_buffer_alloc_latency_metric << (MonotonicMicros() -
allocation_start_us); }};
+ Status injected_status;
+
TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::allocate_tracked_buffer:inject_failure",
+ &injected_status);
+ if (!injected_status.ok()) {
+ *_buffer_alloc_fail_metric << 1;
+ return injected_status;
+ }
+ SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_mem_tracker);
+ Status status = Status::OK();
+ try {
+ *buffer = AsyncCacheWriteBufferPtr(new AsyncCacheWriteBuffer(size,
_mem_tracker));
+ } catch (const std::exception& e) {
+ status = Status::MemoryAllocFailed("allocate async file cache write
buffer failed: {}",
+ e.what());
+ }
+ if (!status.ok()) {
+ *_buffer_alloc_fail_metric << 1;
+ }
+ return status;
+}
+
+void AsyncCacheWriteService::_process_task(AsyncCacheWriteTask task) {
+ Defer finish {[&]() { _finish_active_task(std::move(task)); }};
+
+ const int64_t age_us = MonotonicMicros() - task.submit_ts_us;
+ *_queue_wait_latency_metric << age_us;
+ if (!is_current_write_epoch(task.write_epoch)) {
+ *_drop_stale_epoch_metric << 1;
+ return;
+ }
+
+ const int64_t start_us = MonotonicMicros();
+ Status status = _write_one(task);
Review Comment:
[P1] Contain allocation failures inside each worker task. `_write_one()`
reaches the allocating `get_or_set()` miss path, but this call chain has no
exception boundary; neither `ThreadPool::dispatch_thread()` nor
`Thread::supervise_thread()` catches an exception escaping the runnable. An
actual Doris allocator failure or `std::bad_alloc` here therefore crosses the
pthread entry and terminates the BE instead of dropping one best-effort cache
write. Please convert exceptions to a logged non-OK `Status` around each task
and verify a later task and shutdown still complete after an injected failure.
##########
regression-test/suites/cloud_p0/cache/test_async_file_cache_write.groovy:
##########
@@ -0,0 +1,140 @@
+// 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.
+
+import org.apache.doris.regression.suite.ClusterOptions
+
+suite("test_async_file_cache_write", "docker") {
+ def options = new ClusterOptions()
+ options.cloudMode = true
+ options.setFeNum(1)
+ options.setBeNum(1)
+ options.msNum = 1
+ options.beConfigs += [
Review Comment:
[P1] Disable the S3-writer cache path for this cold-read test. Its new
test-only switch defaults to true, and with asynchronous flush
`UploadFileBuffer::on_upload()` signals writer completion before it calls
`upload_to_local_file_cache()`. Thus `INSERT`/`SYNC` can finish, this suite can
clear the cache, and the delayed load-side fill can then repopulate it before
the first SELECT. The `submittedAfter > submittedBefore` assertion becomes
timing-dependent and no longer proves query-side async population. Add
`enable_file_cache_write_from_s3_file_writer=false` here (or explicitly fence
that writer) before relying on the cold-query metrics.
##########
be/src/io/cache/async_cache_write_service.cpp:
##########
@@ -0,0 +1,731 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "io/cache/async_cache_write_service.h"
+
+#include <algorithm>
+#include <exception>
+#include <limits>
+#include <optional>
+#include <thread>
+#include <type_traits>
+#include <utility>
+
+#include "common/logging.h"
+#include "core/allocator.h"
+#include "cpp/sync_point.h"
+#include "io/cache/block_file_cache.h"
+#include "runtime/thread_context.h"
+#include "util/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
+
+Status resolve_async_file_cache_write_max_pending_bytes_per_disk(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 = 512LL * 1024 * 1024;
+ *resolved_bytes = static_cast<size_t>(std::max(kMinimumAutoPendingBytes,
be_mem_limit / 100));
+ return Status::OK();
+}
+
+class AsyncCacheWriteService::Worker : public
std::enable_shared_from_this<Worker> {
+public:
+ explicit Worker(AsyncCacheWriteService& service) : _service(service) {}
+
+ Status start() {
+ auto self = shared_from_this();
+ return _service._worker_pool->submit_func([self = std::move(self)]() {
self->_run(); });
+ }
+
+ void request_stop() { _stop_requested.store(true,
std::memory_order_release); }
+
+ void wait_until_stopped() { _stopped.wait(); }
+
+private:
+ void _run() {
+ _service._running_worker_count.fetch_add(1, std::memory_order_relaxed);
+ Defer mark_finished {[this]() {
+ const size_t old_running =
+ _service._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 (_service._try_take_task(&task)) {
+ _service._process_task(std::move(task));
+ continue;
+ }
+
+ if (_service._shutdown_requested.load(std::memory_order_acquire) &&
+ _service._pending_count.load(std::memory_order_acquire) == 0) {
+ return;
+ }
+ std::unique_lock lock(_service._queue_mutex);
+ _service._queue_cv.wait(lock, [this]() {
+ const bool shutdown_requested =
+
_service._shutdown_requested.load(std::memory_order_acquire);
+ return !_service._queue.empty() ||
+ _stop_requested.load(std::memory_order_acquire) ||
+ (shutdown_requested &&
+
_service._pending_count.load(std::memory_order_relaxed) == 0);
+ });
+ }
+ }
+
+ AsyncCacheWriteService& _service;
+ 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);
+}
+
+AsyncCacheWriteService::AsyncCacheWriteService(BlockFileCache* cache,
+ AsyncCacheWriteServiceOptions
options)
+ : _cache(cache),
+ _options(std::make_shared<const
AsyncCacheWriteServiceOptions>(options)),
+ _configured_worker_count(options.worker_count) {
+ DORIS_CHECK(_cache != nullptr);
+ DORIS_CHECK(options.worker_count > 0);
+ DORIS_CHECK(options.max_pending_bytes > 0);
+
+ const char* prefix = _cache->get_base_path().c_str();
+ _mem_tracker = MemTrackerLimiter::create_shared(
+ MemTrackerLimiter::Type::CACHE,
+ fmt::format("AsyncFileCacheWrite:{}", _cache->get_base_path()));
+ _pending_count_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_pending_count",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->pending_count();
+ },
+ this);
+ _pending_bytes_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_pending_bytes",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->pending_bytes();
+ },
+ this);
+ _queued_count_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_queue_size",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->queued_count();
+ },
+ this);
+ _queued_bytes_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_queued_bytes",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->queued_bytes();
+ },
+ this);
+ _active_task_count_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_active_tasks",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->active_task_count();
+ },
+ this);
+ _active_bytes_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_active_bytes",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->active_bytes();
+ },
+ this);
+ _running_worker_count_metric =
std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_running_workers",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->running_worker_count();
+ },
+ this);
+ _configured_worker_count_metric =
std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_configured_workers",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->_configured_worker_count.load(
+ std::memory_order_relaxed);
+ },
+ this);
+ _max_pending_bytes_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_max_pending_bytes",
+ [](void* service) {
+ return static_cast<AsyncCacheWriteService*>(service)
+ ->_options.load(std::memory_order_acquire)
+ ->max_pending_bytes;
+ },
+ this);
+ _active_get_or_set_count_metric =
std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_active_get_or_set",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->_active_get_or_set_count.load(
+ std::memory_order_relaxed);
+ },
+ this);
+ _active_append_count_metric =
std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_active_append",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->_active_append_count.load(
+ std::memory_order_relaxed);
+ },
+ this);
+ _active_finalize_count_metric =
std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_active_finalize",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->_active_finalize_count.load(
+ std::memory_order_relaxed);
+ },
+ this);
+ _buffer_memory_metric = std::make_shared<bvar::PassiveStatus<int64_t>>(
+ prefix, "async_cache_write_buffer_memory_bytes",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->buffer_memory_bytes();
+ },
+ this);
+ _submitted_metric =
+ std::make_shared<bvar::Adder<uint64_t>>(prefix,
"async_cache_write_submitted_total");
+ _submitted_bytes_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_submitted_bytes_total");
+ _finished_metric =
+ std::make_shared<bvar::Adder<uint64_t>>(prefix,
"async_cache_write_finished_total");
+ _finished_bytes_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_finished_bytes_total");
+ _worker_finished_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_worker_finished_total");
+ _worker_finished_bytes_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_worker_finished_bytes_total");
+ _evicted_oldest_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_evicted_oldest_total");
+ _evicted_oldest_bytes_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_evicted_oldest_bytes_total");
+ _evicted_oldest_age_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_evicted_oldest_age_us");
+ _rejected_metric =
+ std::make_shared<bvar::Adder<uint64_t>>(prefix,
"async_cache_write_rejected_total");
+ _reject_not_running_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_reject_not_running_total");
+ _reject_backpressure_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_reject_backpressure_total");
+ _buffer_alloc_fail_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_buffer_alloc_fail_total");
+ _submit_latency_metric =
+ std::make_shared<bvar::LatencyRecorder>(prefix,
"async_cache_write_submit_latency_us");
+ _buffer_alloc_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_buffer_alloc_latency_us");
+ _queue_wait_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_queue_wait_latency_us");
+ _queue_lock_wait_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_queue_lock_wait_latency_us");
+ _queue_lock_hold_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_queue_lock_hold_latency_us");
+ _worker_task_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_worker_task_latency_us");
+ _get_or_set_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_get_or_set_latency_us");
+ _append_latency_metric =
+ std::make_shared<bvar::LatencyRecorder>(prefix,
"async_cache_write_append_latency_us");
+ _finalize_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_finalize_latency_us");
+ _skip_downloaded_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_skip_downloaded_total");
+ _skip_downloading_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_skip_downloading_total");
+ _skip_partial_overlap_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_skip_partial_overlap_total");
+ _drop_stale_epoch_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_drop_stale_epoch_total");
+ _skip_deleting_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_skip_deleting_total");
+ _append_fail_metric =
+ std::make_shared<bvar::Adder<uint64_t>>(prefix,
"async_cache_write_append_fail_total");
+ _finalize_fail_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_finalize_fail_total");
+ _persisted_blocks_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_persisted_blocks_total");
+ _persisted_bytes_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_persisted_bytes_total");
+}
+
+AsyncCacheWriteService::~AsyncCacheWriteService() {
+ shutdown();
+}
+
+Status AsyncCacheWriteService::start() {
+ std::lock_guard resize_lock(_resize_mutex);
+ if (_shutdown_requested.load(std::memory_order_acquire) ||
+ !_accepting.load(std::memory_order_acquire)) {
+ return Status::InternalError("async file cache write service is
shutting down");
+ }
+ if (_started.load(std::memory_order_acquire)) {
+ return Status::OK();
+ }
+
+ const size_t worker_count =
_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 AsyncCacheWriteService::try_submit(AsyncCacheWriteTask task) {
+ DORIS_CHECK(task.buffer != nullptr);
+ DORIS_CHECK(task.write_size > 0);
+ DORIS_CHECK(task.write_size <= task.buffer->size());
+ DORIS_CHECK(task.write_size <= std::numeric_limits<size_t>::max() -
task.file_offset);
+ const int64_t submit_start_us = MonotonicMicros();
+ Defer record_submit_latency {
+ [&]() { *_submit_latency_metric << (MonotonicMicros() -
submit_start_us); }};
+ _active_submitters.fetch_add(1, std::memory_order_acq_rel);
+ Defer submitter_done {[&]() { _active_submitters.fetch_sub(1,
std::memory_order_acq_rel); }};
+
TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::try_submit:after_register",
&task);
+ if (!_started.load(std::memory_order_acquire) ||
!_accepting.load(std::memory_order_acquire)) {
+ *_rejected_metric << 1;
+ *_reject_not_running_metric << 1;
+ return false;
+ }
+
+ const size_t task_buffer_bytes = task.buffer->size();
+ std::optional<AsyncCacheWriteTask> victim;
+ {
+ TimedQueueLock lock(_queue_mutex, *_queue_lock_wait_latency_metric,
+ *_queue_lock_hold_latency_metric);
+ const auto options = _options.load(std::memory_order_acquire);
+ const size_t max_pending_bytes = options->max_pending_bytes;
+ const size_t pending_bytes =
_pending_bytes.load(std::memory_order_relaxed);
+ if (_task_buffer_size == 0) {
+ _task_buffer_size = task_buffer_bytes;
+ }
+ DORIS_CHECK(task_buffer_bytes == _task_buffer_size);
+
+ if (task_buffer_bytes > max_pending_bytes) {
+ *_rejected_metric << 1;
+ *_reject_backpressure_metric << 1;
+ return false;
+ }
+
+ const bool has_capacity = pending_bytes <= max_pending_bytes -
task_buffer_bytes;
+ if (!has_capacity && _queue.empty()) {
+ *_rejected_metric << 1;
+ *_reject_backpressure_metric << 1;
+ return false;
+ }
+
+ _queue.push_back(std::move(task));
+ if (has_capacity) {
+ _queued_bytes.fetch_add(task_buffer_bytes,
std::memory_order_relaxed);
+ _pending_count.fetch_add(1, std::memory_order_relaxed);
+ _pending_bytes.fetch_add(task_buffer_bytes,
std::memory_order_relaxed);
+ } else {
+ victim.emplace(std::move(_queue.front()));
+ _queue.pop_front();
+ }
+ }
+
+ *_submitted_metric << 1;
+ *_submitted_bytes_metric << task_buffer_bytes;
+ _queue_cv.notify_one();
+ if (victim) {
+ _finalize_task(std::move(*victim),
TaskFinalizationReason::EVICTED_OLDEST);
+ }
+ return true;
+}
+
+Status AsyncCacheWriteService::allocate_tracked_buffer(size_t size,
+
AsyncCacheWriteBufferPtr* buffer) {
+ DORIS_CHECK(buffer != nullptr);
+ DORIS_CHECK(size > 0);
+ const int64_t allocation_start_us = MonotonicMicros();
+ Defer record_allocation_latency {
+ [&]() { *_buffer_alloc_latency_metric << (MonotonicMicros() -
allocation_start_us); }};
+ Status injected_status;
+
TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::allocate_tracked_buffer:inject_failure",
+ &injected_status);
+ if (!injected_status.ok()) {
+ *_buffer_alloc_fail_metric << 1;
+ return injected_status;
+ }
+ SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_mem_tracker);
+ Status status = Status::OK();
+ try {
+ *buffer = AsyncCacheWriteBufferPtr(new AsyncCacheWriteBuffer(size,
_mem_tracker));
+ } catch (const std::exception& e) {
+ status = Status::MemoryAllocFailed("allocate async file cache write
buffer failed: {}",
+ e.what());
+ }
+ if (!status.ok()) {
+ *_buffer_alloc_fail_metric << 1;
+ }
+ return status;
+}
+
+void AsyncCacheWriteService::_process_task(AsyncCacheWriteTask task) {
+ Defer finish {[&]() { _finish_active_task(std::move(task)); }};
+
+ const int64_t age_us = MonotonicMicros() - task.submit_ts_us;
+ *_queue_wait_latency_metric << age_us;
+ if (!is_current_write_epoch(task.write_epoch)) {
+ *_drop_stale_epoch_metric << 1;
+ return;
+ }
+
+ const int64_t start_us = MonotonicMicros();
+ Status status = _write_one(task);
+ *_worker_task_latency_metric << (MonotonicMicros() - start_us);
+ if (!status.ok()) {
+ LOG(WARNING) << "Async file cache write failed, cache=" <<
_cache->get_base_path()
+ << ", hash=" << task.cache_hash.to_string() << ",
offset=" << task.file_offset
+ << ", size=" << task.write_size << ", status=" << status;
+ }
+}
+
+bool AsyncCacheWriteService::_try_take_task(AsyncCacheWriteTask* task) {
+ TimedQueueLock lock(_queue_mutex, *_queue_lock_wait_latency_metric,
+ *_queue_lock_hold_latency_metric);
+ if (_queue.empty()) {
+ return false;
+ }
+
+ *task = std::move(_queue.front());
+ _queue.pop_front();
+ const size_t task_buffer_bytes = task->buffer->size();
+ _queued_bytes.fetch_sub(task_buffer_bytes, std::memory_order_relaxed);
+ _active_task_count.fetch_add(1, std::memory_order_relaxed);
+ _active_bytes.fetch_add(task_buffer_bytes, std::memory_order_relaxed);
+ return true;
+}
+
+Status AsyncCacheWriteService::_write_one(const AsyncCacheWriteTask& task) {
+ if (!is_current_write_epoch(task.write_epoch)) {
+ *_drop_stale_epoch_metric << 1;
+ return Status::OK();
+ }
+
+ ReadStatistics dummy_stats;
+ CacheContext context;
+ context.query_id = task.admission_ctx.query_id;
+ context.cache_type = task.admission_ctx.cache_type;
+ context.expiration_time = task.admission_ctx.expiration_time;
+ context.tablet_id = task.admission_ctx.tablet_id;
+ context.is_warmup = task.admission_ctx.is_warmup;
+ context.stats = &dummy_stats;
+ auto holder = [&]() {
+ ScopedActiveCounter active_get_or_set(_active_get_or_set_count);
+ const int64_t start_us = MonotonicMicros();
+ Defer record_latency {
+ [&]() { *_get_or_set_latency_metric << (MonotonicMicros() -
start_us); }};
+
TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::_write_one:before_get_or_set",
&task);
+ auto result =
+ _cache->get_or_set(task.cache_hash, task.file_offset,
task.write_size, context);
+
TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::_write_one:after_get_or_set",
&task);
+ return result;
+ }();
+
+ if (!is_current_write_epoch(task.write_epoch)) {
+ *_drop_stale_epoch_metric << 1;
+ return Status::OK();
+ }
+
+ const size_t task_end = task.file_offset + task.write_size;
+ for (const auto& block : holder.file_blocks) {
+ if (block->range().left < task.file_offset || block->range().right >=
task_end) {
+ *_skip_partial_overlap_metric << 1;
+ continue;
+ }
+ if (!is_current_write_epoch(task.write_epoch)) {
+ *_drop_stale_epoch_metric << 1;
+ return Status::OK();
+ }
+ if (_cache->is_block_deleting(block)) {
+ *_skip_deleting_metric << 1;
+ continue;
+ }
+
+ switch (block->state()) {
+ case FileBlock::State::DOWNLOADED:
+ *_skip_downloaded_metric << 1;
+ continue;
+ case FileBlock::State::DOWNLOADING:
+ *_skip_downloading_metric << 1;
+ continue;
+ case FileBlock::State::SKIP_CACHE:
+ continue;
+ case FileBlock::State::EMPTY:
+ break;
+ }
+
+ if (block->get_or_set_downloader() != FileBlock::get_caller_id()) {
+ *_skip_downloading_metric << 1;
+ continue;
+ }
+ const size_t buffer_offset = block->range().left - task.file_offset;
+ DORIS_CHECK(buffer_offset <= task.write_size);
+ DORIS_CHECK(block->range().size() <= task.write_size - buffer_offset);
+ Status status;
+ {
+ ScopedActiveCounter active_append(_active_append_count);
+
TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::_write_one:before_append",
&task);
+ const int64_t start_us = MonotonicMicros();
+ status = block->append(
+ Slice(task.buffer->data() + buffer_offset,
block->range().size()));
+ *_append_latency_metric << (MonotonicMicros() - start_us);
+ }
+ if (!status.ok()) {
+ *_append_fail_metric << 1;
+ LOG(WARNING) << "Append async file cache block failed, cache="
+ << _cache->get_base_path() << ", hash=" <<
task.cache_hash.to_string()
+ << ", offset=" << block->offset() << ", size=" <<
block->range().size()
+ << ", status=" << status;
+ continue;
+ }
+ {
+ ScopedActiveCounter active_finalize(_active_finalize_count);
+ const int64_t start_us = MonotonicMicros();
+ status = block->finalize();
+ *_finalize_latency_metric << (MonotonicMicros() - start_us);
+ }
+ if (!status.ok()) {
+ *_finalize_fail_metric << 1;
+ LOG(WARNING) << "Finalize async file cache block failed, cache="
+ << _cache->get_base_path() << ", hash=" <<
task.cache_hash.to_string()
+ << ", offset=" << block->offset() << ", size=" <<
block->range().size()
+ << ", status=" << status;
+ continue;
+ }
+ *_persisted_blocks_metric << 1;
+ *_persisted_bytes_metric << block->range().size();
+ }
+ return Status::OK();
+}
+
+void AsyncCacheWriteService::_finish_active_task(AsyncCacheWriteTask task) {
+ const size_t task_buffer_bytes = task.buffer->size();
+ bool became_empty = false;
+ {
+ TimedQueueLock lock(_queue_mutex, *_queue_lock_wait_latency_metric,
+ *_queue_lock_hold_latency_metric);
+ const size_t old_active = _active_task_count.fetch_sub(1,
std::memory_order_relaxed);
+ DCHECK_GT(old_active, 0);
+ _active_bytes.fetch_sub(task_buffer_bytes, std::memory_order_relaxed);
+ const size_t old_pending = _pending_count.fetch_sub(1,
std::memory_order_relaxed);
Review Comment:
[P1] Keep finalizing buffers inside the pending quota. This subtracts the
task before `_finalize_task()` runs the callback that removes its inflight
entry and before the buffer is released. The full-queue path has the same gap:
it transfers a counted slot to the new task, then finalizes the displaced
victim outside the lock, so concurrent producers can each retain an additional
uncounted full-block buffer while the configured limit appears satisfied.
`pending_count` can also reach zero while an inflight entry remains, making the
new regression's pending-only drain race its disk-probe assertion. Account
reserved/finalizing bytes through callback and buffer release, and make drain
users wait on that same completion invariant.
##########
be/src/io/cache/async_cache_write_service.cpp:
##########
@@ -0,0 +1,731 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "io/cache/async_cache_write_service.h"
+
+#include <algorithm>
+#include <exception>
+#include <limits>
+#include <optional>
+#include <thread>
+#include <type_traits>
+#include <utility>
+
+#include "common/logging.h"
+#include "core/allocator.h"
+#include "cpp/sync_point.h"
+#include "io/cache/block_file_cache.h"
+#include "runtime/thread_context.h"
+#include "util/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
+
+Status resolve_async_file_cache_write_max_pending_bytes_per_disk(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 = 512LL * 1024 * 1024;
+ *resolved_bytes = static_cast<size_t>(std::max(kMinimumAutoPendingBytes,
be_mem_limit / 100));
+ return Status::OK();
+}
+
+class AsyncCacheWriteService::Worker : public
std::enable_shared_from_this<Worker> {
+public:
+ explicit Worker(AsyncCacheWriteService& service) : _service(service) {}
+
+ Status start() {
+ auto self = shared_from_this();
+ return _service._worker_pool->submit_func([self = std::move(self)]() {
self->_run(); });
+ }
+
+ void request_stop() { _stop_requested.store(true,
std::memory_order_release); }
+
+ void wait_until_stopped() { _stopped.wait(); }
+
+private:
+ void _run() {
+ _service._running_worker_count.fetch_add(1, std::memory_order_relaxed);
+ Defer mark_finished {[this]() {
+ const size_t old_running =
+ _service._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 (_service._try_take_task(&task)) {
+ _service._process_task(std::move(task));
+ continue;
+ }
+
+ if (_service._shutdown_requested.load(std::memory_order_acquire) &&
+ _service._pending_count.load(std::memory_order_acquire) == 0) {
+ return;
+ }
+ std::unique_lock lock(_service._queue_mutex);
+ _service._queue_cv.wait(lock, [this]() {
+ const bool shutdown_requested =
+
_service._shutdown_requested.load(std::memory_order_acquire);
+ return !_service._queue.empty() ||
+ _stop_requested.load(std::memory_order_acquire) ||
+ (shutdown_requested &&
+
_service._pending_count.load(std::memory_order_relaxed) == 0);
+ });
+ }
+ }
+
+ AsyncCacheWriteService& _service;
+ 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);
+}
+
+AsyncCacheWriteService::AsyncCacheWriteService(BlockFileCache* cache,
+ AsyncCacheWriteServiceOptions
options)
+ : _cache(cache),
+ _options(std::make_shared<const
AsyncCacheWriteServiceOptions>(options)),
+ _configured_worker_count(options.worker_count) {
+ DORIS_CHECK(_cache != nullptr);
+ DORIS_CHECK(options.worker_count > 0);
+ DORIS_CHECK(options.max_pending_bytes > 0);
+
+ const char* prefix = _cache->get_base_path().c_str();
+ _mem_tracker = MemTrackerLimiter::create_shared(
+ MemTrackerLimiter::Type::CACHE,
+ fmt::format("AsyncFileCacheWrite:{}", _cache->get_base_path()));
+ _pending_count_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_pending_count",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->pending_count();
+ },
+ this);
+ _pending_bytes_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_pending_bytes",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->pending_bytes();
+ },
+ this);
+ _queued_count_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_queue_size",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->queued_count();
+ },
+ this);
+ _queued_bytes_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_queued_bytes",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->queued_bytes();
+ },
+ this);
+ _active_task_count_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_active_tasks",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->active_task_count();
+ },
+ this);
+ _active_bytes_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_active_bytes",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->active_bytes();
+ },
+ this);
+ _running_worker_count_metric =
std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_running_workers",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->running_worker_count();
+ },
+ this);
+ _configured_worker_count_metric =
std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_configured_workers",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->_configured_worker_count.load(
+ std::memory_order_relaxed);
+ },
+ this);
+ _max_pending_bytes_metric = std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_max_pending_bytes",
+ [](void* service) {
+ return static_cast<AsyncCacheWriteService*>(service)
+ ->_options.load(std::memory_order_acquire)
+ ->max_pending_bytes;
+ },
+ this);
+ _active_get_or_set_count_metric =
std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_active_get_or_set",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->_active_get_or_set_count.load(
+ std::memory_order_relaxed);
+ },
+ this);
+ _active_append_count_metric =
std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_active_append",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->_active_append_count.load(
+ std::memory_order_relaxed);
+ },
+ this);
+ _active_finalize_count_metric =
std::make_shared<bvar::PassiveStatus<size_t>>(
+ prefix, "async_cache_write_active_finalize",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->_active_finalize_count.load(
+ std::memory_order_relaxed);
+ },
+ this);
+ _buffer_memory_metric = std::make_shared<bvar::PassiveStatus<int64_t>>(
+ prefix, "async_cache_write_buffer_memory_bytes",
+ [](void* service) {
+ return
static_cast<AsyncCacheWriteService*>(service)->buffer_memory_bytes();
+ },
+ this);
+ _submitted_metric =
+ std::make_shared<bvar::Adder<uint64_t>>(prefix,
"async_cache_write_submitted_total");
+ _submitted_bytes_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_submitted_bytes_total");
+ _finished_metric =
+ std::make_shared<bvar::Adder<uint64_t>>(prefix,
"async_cache_write_finished_total");
+ _finished_bytes_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_finished_bytes_total");
+ _worker_finished_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_worker_finished_total");
+ _worker_finished_bytes_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_worker_finished_bytes_total");
+ _evicted_oldest_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_evicted_oldest_total");
+ _evicted_oldest_bytes_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_evicted_oldest_bytes_total");
+ _evicted_oldest_age_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_evicted_oldest_age_us");
+ _rejected_metric =
+ std::make_shared<bvar::Adder<uint64_t>>(prefix,
"async_cache_write_rejected_total");
+ _reject_not_running_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_reject_not_running_total");
+ _reject_backpressure_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_reject_backpressure_total");
+ _buffer_alloc_fail_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_buffer_alloc_fail_total");
+ _submit_latency_metric =
+ std::make_shared<bvar::LatencyRecorder>(prefix,
"async_cache_write_submit_latency_us");
+ _buffer_alloc_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_buffer_alloc_latency_us");
+ _queue_wait_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_queue_wait_latency_us");
+ _queue_lock_wait_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_queue_lock_wait_latency_us");
+ _queue_lock_hold_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_queue_lock_hold_latency_us");
+ _worker_task_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_worker_task_latency_us");
+ _get_or_set_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_get_or_set_latency_us");
+ _append_latency_metric =
+ std::make_shared<bvar::LatencyRecorder>(prefix,
"async_cache_write_append_latency_us");
+ _finalize_latency_metric = std::make_shared<bvar::LatencyRecorder>(
+ prefix, "async_cache_write_finalize_latency_us");
+ _skip_downloaded_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_skip_downloaded_total");
+ _skip_downloading_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_skip_downloading_total");
+ _skip_partial_overlap_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_skip_partial_overlap_total");
+ _drop_stale_epoch_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_drop_stale_epoch_total");
+ _skip_deleting_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_skip_deleting_total");
+ _append_fail_metric =
+ std::make_shared<bvar::Adder<uint64_t>>(prefix,
"async_cache_write_append_fail_total");
+ _finalize_fail_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_finalize_fail_total");
+ _persisted_blocks_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_persisted_blocks_total");
+ _persisted_bytes_metric = std::make_shared<bvar::Adder<uint64_t>>(
+ prefix, "async_cache_write_persisted_bytes_total");
+}
+
+AsyncCacheWriteService::~AsyncCacheWriteService() {
+ shutdown();
+}
+
+Status AsyncCacheWriteService::start() {
+ std::lock_guard resize_lock(_resize_mutex);
+ if (_shutdown_requested.load(std::memory_order_acquire) ||
+ !_accepting.load(std::memory_order_acquire)) {
+ return Status::InternalError("async file cache write service is
shutting down");
+ }
+ if (_started.load(std::memory_order_acquire)) {
+ return Status::OK();
+ }
+
+ const size_t worker_count =
_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 AsyncCacheWriteService::try_submit(AsyncCacheWriteTask task) {
+ DORIS_CHECK(task.buffer != nullptr);
+ DORIS_CHECK(task.write_size > 0);
+ DORIS_CHECK(task.write_size <= task.buffer->size());
+ DORIS_CHECK(task.write_size <= std::numeric_limits<size_t>::max() -
task.file_offset);
+ const int64_t submit_start_us = MonotonicMicros();
+ Defer record_submit_latency {
+ [&]() { *_submit_latency_metric << (MonotonicMicros() -
submit_start_us); }};
+ _active_submitters.fetch_add(1, std::memory_order_acq_rel);
+ Defer submitter_done {[&]() { _active_submitters.fetch_sub(1,
std::memory_order_acq_rel); }};
+
TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::try_submit:after_register",
&task);
+ if (!_started.load(std::memory_order_acquire) ||
!_accepting.load(std::memory_order_acquire)) {
+ *_rejected_metric << 1;
+ *_reject_not_running_metric << 1;
+ return false;
+ }
+
+ const size_t task_buffer_bytes = task.buffer->size();
+ std::optional<AsyncCacheWriteTask> victim;
+ {
+ TimedQueueLock lock(_queue_mutex, *_queue_lock_wait_latency_metric,
+ *_queue_lock_hold_latency_metric);
+ const auto options = _options.load(std::memory_order_acquire);
+ const size_t max_pending_bytes = options->max_pending_bytes;
+ const size_t pending_bytes =
_pending_bytes.load(std::memory_order_relaxed);
+ if (_task_buffer_size == 0) {
+ _task_buffer_size = task_buffer_bytes;
+ }
+ DORIS_CHECK(task_buffer_bytes == _task_buffer_size);
+
+ if (task_buffer_bytes > max_pending_bytes) {
+ *_rejected_metric << 1;
+ *_reject_backpressure_metric << 1;
+ return false;
+ }
+
+ const bool has_capacity = pending_bytes <= max_pending_bytes -
task_buffer_bytes;
+ if (!has_capacity && _queue.empty()) {
+ *_rejected_metric << 1;
+ *_reject_backpressure_metric << 1;
+ return false;
+ }
+
+ _queue.push_back(std::move(task));
+ if (has_capacity) {
+ _queued_bytes.fetch_add(task_buffer_bytes,
std::memory_order_relaxed);
+ _pending_count.fetch_add(1, std::memory_order_relaxed);
+ _pending_bytes.fetch_add(task_buffer_bytes,
std::memory_order_relaxed);
+ } else {
+ victim.emplace(std::move(_queue.front()));
+ _queue.pop_front();
+ }
+ }
+
+ *_submitted_metric << 1;
+ *_submitted_bytes_metric << task_buffer_bytes;
+ _queue_cv.notify_one();
+ if (victim) {
+ _finalize_task(std::move(*victim),
TaskFinalizationReason::EVICTED_OLDEST);
+ }
+ return true;
+}
+
+Status AsyncCacheWriteService::allocate_tracked_buffer(size_t size,
+
AsyncCacheWriteBufferPtr* buffer) {
+ DORIS_CHECK(buffer != nullptr);
+ DORIS_CHECK(size > 0);
+ const int64_t allocation_start_us = MonotonicMicros();
+ Defer record_allocation_latency {
+ [&]() { *_buffer_alloc_latency_metric << (MonotonicMicros() -
allocation_start_us); }};
+ Status injected_status;
+
TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::allocate_tracked_buffer:inject_failure",
+ &injected_status);
+ if (!injected_status.ok()) {
+ *_buffer_alloc_fail_metric << 1;
+ return injected_status;
+ }
+ SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_mem_tracker);
Review Comment:
[P1] Enable the catchable allocation path here. Switching to `_mem_tracker`
does not set `enable_thread_catch_bad_alloc`; both allocator system-memory and
tracker-limit checks only throw `MEM_ALLOC_FAILED` when that flag is set, and
otherwise log that allocation will continue. Consequently this `try`/non-OK
fallback cannot apply backpressure at Doris's memory limits and a best-effort
cache fill can allocate past them until the OS allocation itself fails. Please
use the standard `RETURN_IF_CATCH_EXCEPTION` boundary around construction
(while preserving the failure metric) and test the real allocator-limit path.
##########
be/src/io/cache/benchmark/async_file_cache_write_microbench.cpp:
##########
@@ -0,0 +1,1099 @@
+// 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_common.h"
+
+#if defined(BE_TEST) && defined(BUILD_FILE_CACHE_MICROBENCH_TOOL)
+
+#include <gflags/gflags.h>
+#include <glog/logging.h>
+
+#include <algorithm>
+#include <atomic>
+#include <barrier>
+#include <chrono>
+#include <cmath>
+#include <cstddef>
+#include <cstdint>
+#include <cstdlib>
+#include <cstring>
+#include <filesystem>
+#include <iomanip>
+#include <iostream>
+#include <memory>
+#include <mutex>
+#include <numeric>
+#include <sstream>
+#include <string>
+#include <string_view>
+#include <thread>
+#include <utility>
+#include <vector>
+
+#include "cloud/config.h"
+#include "common/config.h"
+#include "common/status.h"
+#include "io/cache/async_cache_write_service.h"
+#include "io/cache/block_file_cache.h"
+#include "io/cache/block_file_cache_factory.h"
+#include "io/cache/cached_remote_file_reader.h"
+#include "io/cache/fs_file_cache_storage.h"
+#include "io/cache/inflight_write_buffer_index.h"
+#include "io/fs/file_reader.h"
+#include "io/fs/path.h"
+#include "runtime/exec_env.h"
+#include "runtime/thread_context.h"
+#include "util/cpu_info.h"
+#include "util/disk_info.h"
+#include "util/mem_info.h"
+#include "util/slice.h"
+#include "util/time.h"
+
+DEFINE_string(benchmark_mode, "all",
+ "Comma-separated benchmark groups: reader, service, index, or
all");
+DEFINE_string(cache_path, "./output/async_file_cache_write_microbench",
+ "Directory used by the real filesystem-backed BlockFileCache");
+DEFINE_uint64(block_size, 1024 * 1024,
+ "File-cache block size and stride used by cold reader misses");
+DEFINE_uint64(request_size, 64 * 1024, "Bytes returned to the caller by each
reader operation");
+DEFINE_uint64(reader_operations, 128, "Cold read operations in each sync/async
reader case");
+DEFINE_uint64(service_task_size, 1024 * 1024, "Payload bytes in each direct
service task");
+DEFINE_uint64(service_operations, 256, "Attempted tasks in each direct service
case");
+DEFINE_uint64(service_key_count, 64, "Logical remote files spread across
direct service tasks");
+DEFINE_uint64(index_operations_per_thread, 100000,
+ "Inflight-index lookups performed by each producer thread");
+DEFINE_uint64(index_key_count, 4096, "Keys used by the representative sharded
index case");
+DEFINE_int32(producer_threads, 16, "Concurrent foreground readers or task
producers");
+DEFINE_int32(reader_workers, 16, "Async write workers used by the reader
comparison");
+DEFINE_string(worker_counts, "1,4,16",
+ "Comma-separated async write worker counts used by service
scaling cases");
+DEFINE_uint64(repetitions, 5, "Measured repetitions of every selected
benchmark case");
+DEFINE_uint64(backpressure_pending_bytes, 64 * 1024 * 1024,
+ "Pending-buffer byte limit used by the saturated service case");
+DEFINE_uint64(queue_sample_interval_us, 50,
+ "Sampling interval for pending, queued, and inflight peak
values");
+DEFINE_uint64(timeout_seconds, 120, "Maximum time to wait for one benchmark
case to drain");
+DEFINE_bool(keep_cache, false, "Keep benchmark cache files after the process
exits");
+
+namespace doris::io {
+namespace {
+
+using Clock = std::chrono::steady_clock;
+using Nanoseconds = std::chrono::nanoseconds;
+
+constexpr size_t kMiB = 1024 * 1024;
+
+/// Compact percentile summary for foreground operation latency.
+struct LatencySummary {
+ double average_us {0};
+ double p50_us {0};
+ double p95_us {0};
+ double p99_us {0};
+ double maximum_us {0};
+};
+
+/// Return one nearest-rank percentile from an already sorted nanosecond
sample set.
+/// @param sorted_ns Ascending latency samples in nanoseconds.
+/// @param percentile Requested percentile in the inclusive range [0, 1].
+double percentile_us(const std::vector<int64_t>& sorted_ns, double percentile)
{
+ DORIS_CHECK(!sorted_ns.empty());
+ DORIS_CHECK(percentile > 0 && percentile <= 1);
+ const size_t index =
+ static_cast<size_t>(std::ceil(percentile *
static_cast<double>(sorted_ns.size()))) - 1;
+ return static_cast<double>(sorted_ns[std::min(index, sorted_ns.size() -
1)]) / 1000.0;
+}
+
+/// Merge per-thread samples and calculate stable latency percentiles.
+/// @param per_thread_ns Independently owned samples, one vector per producer.
+LatencySummary summarize_latencies(const std::vector<std::vector<int64_t>>&
per_thread_ns) {
+ size_t sample_count = 0;
+ for (const auto& samples : per_thread_ns) {
+ sample_count += samples.size();
+ }
+ DORIS_CHECK(sample_count > 0);
+
+ std::vector<int64_t> sorted_ns;
+ sorted_ns.reserve(sample_count);
+ for (const auto& samples : per_thread_ns) {
+ sorted_ns.insert(sorted_ns.end(), samples.begin(), samples.end());
+ }
+ std::sort(sorted_ns.begin(), sorted_ns.end());
+ const int64_t total_ns = std::accumulate(sorted_ns.begin(),
sorted_ns.end(), int64_t {0});
+ return LatencySummary {
+ .average_us =
+ static_cast<double>(total_ns) /
static_cast<double>(sample_count) / 1000.0,
+ .p50_us = percentile_us(sorted_ns, 0.50),
+ .p95_us = percentile_us(sorted_ns, 0.95),
+ .p99_us = percentile_us(sorted_ns, 0.99),
+ .maximum_us = static_cast<double>(sorted_ns.back()) / 1000.0,
+ };
+}
+
+/// Parse a comma-separated positive integer list used for worker scaling.
+/// @param text Raw gflag value such as "1,4,16".
+/// @param values Parsed worker counts in input order.
+Status parse_positive_integer_list(std::string_view text, std::vector<size_t>*
values) {
+ DORIS_CHECK(values != nullptr);
+ values->clear();
+ std::stringstream stream {std::string(text)};
+ std::string token;
+ while (std::getline(stream, token, ',')) {
+ try {
+ const long long value = std::stoll(token);
+ if (value <= 0) {
+ return Status::InvalidArgument("worker count must be positive:
{}", token);
+ }
+ values->push_back(static_cast<size_t>(value));
+ } catch (const std::exception& error) {
+ return Status::InvalidArgument("invalid worker count '{}': {}",
token, error.what());
+ }
+ }
+ if (values->empty()) {
+ return Status::InvalidArgument("worker_counts cannot be empty");
+ }
+ return Status::OK();
+}
+
+/// Split the selected benchmark groups while preserving a small command-line
surface.
+/// @param text Raw mode string.
+std::vector<std::string> parse_modes(std::string_view text) {
+ std::stringstream stream {std::string(text)};
+ std::vector<std::string> modes;
+ std::string mode;
+ while (std::getline(stream, mode, ',')) {
+ if (!mode.empty()) {
+ modes.emplace_back(std::move(mode));
+ }
+ }
+ return modes;
+}
+
+/// Return whether a benchmark group was requested explicitly or through "all".
+/// @param modes Parsed benchmark groups.
+/// @param target Group to test.
+bool mode_enabled(const std::vector<std::string>& modes, std::string_view
target) {
+ return std::find(modes.begin(), modes.end(), "all") != modes.end() ||
+ std::find(modes.begin(), modes.end(), target) != modes.end();
+}
+
+/// Validate sizes and concurrency before allocating cache capacity or
starting threads.
+Status validate_flags(const std::vector<std::string>& modes,
+ const std::vector<size_t>& worker_counts) {
+ if (modes.empty()) {
+ return Status::InvalidArgument("benchmark_mode cannot be empty");
+ }
+ for (const auto& mode : modes) {
+ if (mode != "all" && mode != "reader" && mode != "service" && mode !=
"index") {
+ return Status::InvalidArgument("unsupported benchmark mode: {}",
mode);
+ }
+ }
+ DORIS_CHECK(!worker_counts.empty());
+ if (FLAGS_producer_threads <= 0 || FLAGS_reader_workers <= 0) {
+ return Status::InvalidArgument("producer_threads and reader_workers
must be positive");
+ }
+ if (FLAGS_block_size == 0 || FLAGS_request_size == 0 || FLAGS_request_size
> FLAGS_block_size) {
+ return Status::InvalidArgument("sizes must satisfy 0 < request_size <=
block_size");
+ }
+ if (FLAGS_reader_operations <
static_cast<uint64_t>(FLAGS_producer_threads) ||
+ FLAGS_service_operations <
static_cast<uint64_t>(FLAGS_producer_threads)) {
+ return Status::InvalidArgument(
+ "reader_operations and service_operations must be at least
producer_threads");
+ }
+ if (FLAGS_service_task_size != FLAGS_block_size || FLAGS_service_key_count
== 0 ||
+ FLAGS_index_operations_per_thread == 0 || FLAGS_index_key_count == 0 ||
+ FLAGS_backpressure_pending_bytes == 0 ||
FLAGS_queue_sample_interval_us == 0 ||
+ FLAGS_timeout_seconds == 0 || FLAGS_repetitions == 0) {
+ return Status::InvalidArgument(
+ "operation counts, timeouts, and service_task_size ==
block_size are required");
+ }
+ return Status::OK();
+}
+
+/// Produce deterministic data without adding network or object-store latency
to reader results.
+class SyntheticRemoteFileReader final : public FileReader {
+public:
+ /// @param path Stable logical path used to derive the file-cache hash.
+ /// @param file_size Virtual file length; no payload is allocated for it.
+ /// @param block_size Pattern granularity used to validate copied bytes.
+ SyntheticRemoteFileReader(Path path, size_t file_size, size_t block_size)
+ : _path(std::move(path)), _file_size(file_size),
_block_size(block_size) {}
+
+ Status close() override {
+ _closed.store(true, std::memory_order_release);
+ return Status::OK();
+ }
+
+ const Path& path() const override { return _path; }
+ size_t size() const override { return _file_size; }
+ bool closed() const override { return
_closed.load(std::memory_order_acquire); }
+ int64_t mtime() const override { return 0; }
+
+ /// Return the byte value expected at an aligned benchmark block.
+ /// @param offset File offset inside the virtual source.
+ char expected_byte(size_t offset) const {
+ return static_cast<char>((offset / _block_size) % 251);
+ }
+
+protected:
+ /// Fill the requested span from a deterministic virtual file.
+ Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
+ const IOContext* io_ctx) override {
+ DORIS_CHECK(bytes_read != nullptr);
+ if (offset > _file_size || result.size > _file_size - offset) {
+ return Status::InvalidArgument("synthetic read [{}, {}) exceeds
file size {}", offset,
+ offset + result.size, _file_size);
+ }
+ size_t copied = 0;
+ while (copied < result.size) {
+ const size_t current_offset = offset + copied;
+ const size_t block_end =
+ std::min(_file_size, (current_offset / _block_size + 1) *
_block_size);
+ const size_t bytes = std::min(result.size - copied, block_end -
current_offset);
+ std::memset(result.data + copied, expected_byte(current_offset),
bytes);
+ copied += bytes;
+ }
+ *bytes_read = result.size;
+ return Status::OK();
+ }
+
+private:
+ Path _path;
+ size_t _file_size;
+ size_t _block_size;
+ std::atomic<bool> _closed {false};
+};
+
+/// Record the first worker-thread error without obscuring the performance hot
path.
+class ConcurrentError {
+public:
+ /// Store the first non-OK status observed by any producer.
+ void set(Status status) {
+ if (status.ok()) {
+ return;
+ }
+ bool expected = false;
+ if (_failed.compare_exchange_strong(expected, true,
std::memory_order_acq_rel)) {
+ std::lock_guard lock(_mutex);
+ _status = std::move(status);
+ }
+ }
+
+ bool failed() const { return _failed.load(std::memory_order_acquire); }
+
+ /// Return the stored error after producer threads have joined.
+ Status status() const {
+ std::lock_guard lock(_mutex);
+ return _status;
+ }
+
+private:
+ std::atomic<bool> _failed {false};
+ mutable std::mutex _mutex;
+ Status _status;
+};
+
+/// Sample public queue gauges so short benchmark cases still report a
meaningful high-water mark.
+class QueuePeakSampler {
+public:
+ /// @param service Service whose pending and queued gauges are sampled.
+ /// @param index Inflight index paired with the service.
+ QueuePeakSampler(AsyncCacheWriteService* service,
InflightWriteBufferIndex* index)
+ : _service(service), _index(index) {
+ DORIS_CHECK(_service != nullptr);
+ DORIS_CHECK(_index != nullptr);
+ }
+
+ /// Start the sampling thread. The sampler does not mutate benchmark state.
+ void start() {
+ _running.store(true, std::memory_order_release);
+ _thread = std::thread([this]() {
+ while (_running.load(std::memory_order_acquire)) {
+ sample();
+ std::this_thread::sleep_for(
+
std::chrono::microseconds(FLAGS_queue_sample_interval_us));
+ }
+ sample();
+ });
+ }
+
+ /// Stop sampling and join before reading peak values.
+ void stop() {
+ _running.store(false, std::memory_order_release);
+ if (_thread.joinable()) {
+ _thread.join();
+ }
+ }
+
+ ~QueuePeakSampler() { stop(); }
+
+ size_t peak_pending() const { return
_peak_pending.load(std::memory_order_relaxed); }
+ size_t peak_queued() const { return
_peak_queued.load(std::memory_order_relaxed); }
+ size_t peak_inflight() const { return
_peak_inflight.load(std::memory_order_relaxed); }
+ size_t peak_buffer_bytes() const { return
_peak_buffer_bytes.load(std::memory_order_relaxed); }
+
+private:
+ /// Capture one internally consistent set of independently sampled public
gauges.
+ void sample() {
+ update_max(&_peak_pending, _service->pending_count());
+ update_max(&_peak_queued, _service->queued_count());
+ update_max(&_peak_inflight, _index->count());
+ const int64_t buffer_memory_bytes = _service->buffer_memory_bytes();
+ DORIS_CHECK(buffer_memory_bytes >= 0);
+ update_max(&_peak_buffer_bytes,
static_cast<size_t>(buffer_memory_bytes));
+ }
+
+ /// Atomically preserve the largest sampled gauge value.
+ static void update_max(std::atomic<size_t>* maximum, size_t value) {
+ size_t current = maximum->load(std::memory_order_relaxed);
+ while (current < value &&
+ !maximum->compare_exchange_weak(current, value,
std::memory_order_relaxed)) {
+ }
+ }
+
+ AsyncCacheWriteService* _service;
+ InflightWriteBufferIndex* _index;
+ std::atomic<bool> _running {false};
+ std::thread _thread;
+ std::atomic<size_t> _peak_pending {0};
+ std::atomic<size_t> _peak_queued {0};
+ std::atomic<size_t> _peak_inflight {0};
+ std::atomic<size_t> _peak_buffer_bytes {0};
+};
+
+/// Poll one asynchronous completion predicate with a bounded failure mode.
+/// @param predicate Returns true after the expected state is reached.
+/// @param description Included in timeout diagnostics.
+template <typename Predicate>
+Status wait_until(Predicate&& predicate, std::string_view description) {
+ const auto deadline = Clock::now() +
std::chrono::seconds(FLAGS_timeout_seconds);
+ while (!predicate()) {
+ if (Clock::now() >= deadline) {
+ return Status::TimedOut("timed out waiting for {}", description);
+ }
+ std::this_thread::sleep_for(std::chrono::microseconds(100));
+ }
+ return Status::OK();
+}
+
+/// Own one real filesystem cache and install its factory into the otherwise
minimal ExecEnv.
+class BenchmarkEnvironment {
+public:
+ BenchmarkEnvironment() = default;
+
+ ~BenchmarkEnvironment() {
+ if (_factory) {
+ ExecEnv::GetInstance()->set_file_cache_factory(nullptr);
+ _factory.reset();
+ }
+ if (_owns_cache_path && !FLAGS_keep_cache) {
+ std::error_code error;
+ std::filesystem::remove_all(FLAGS_cache_path, error);
+ }
+ }
+
+ /// Configure globals, create the cache, and wait for its asynchronous
metadata open.
+ /// @param cache_capacity Bytes reserved for the benchmark's normal queue.
+ Status initialize(size_t cache_capacity) {
+ std::error_code error;
+ const auto cache_path =
+ std::filesystem::absolute(FLAGS_cache_path,
error).lexically_normal();
+ if (error) {
+ return Status::IOError("failed to resolve cache path {}: {}",
FLAGS_cache_path,
+ error.message());
+ }
+ const auto current_path = std::filesystem::current_path(error);
+ if (error) {
+ return Status::IOError("failed to resolve current directory: {}",
error.message());
+ }
+ if (cache_path == cache_path.root_path() || cache_path ==
current_path) {
+ return Status::InvalidArgument("cache path must be a dedicated
subdirectory: {}",
+ cache_path.string());
+ }
+ const bool cache_path_exists = std::filesystem::exists(cache_path,
error);
+ if (error) {
+ return Status::IOError("failed to inspect cache path {}: {}",
cache_path.string(),
+ error.message());
+ }
+ if (cache_path_exists) {
+ const bool cache_path_is_directory =
std::filesystem::is_directory(cache_path, error);
+ if (error) {
+ return Status::IOError("failed to inspect cache path {}: {}",
cache_path.string(),
+ error.message());
+ }
+ if (!cache_path_is_directory) {
+ return Status::InvalidArgument("cache path is not a directory:
{}",
+ cache_path.string());
+ }
+ const bool cache_path_is_empty =
std::filesystem::is_empty(cache_path, error);
+ if (error) {
+ return Status::IOError("failed to inspect cache path {}: {}",
cache_path.string(),
+ error.message());
+ }
+ if (!cache_path_is_empty) {
+ return Status::InvalidArgument(
+ "cache path must not exist or must be an empty
directory: {}",
+ cache_path.string());
+ }
+ } else {
+ std::filesystem::create_directories(cache_path, error);
+ if (error) {
+ return Status::IOError("failed to create cache path {}: {}",
cache_path.string(),
+ error.message());
+ }
+ }
+ _owns_cache_path = true;
+
+ config::enable_async_file_cache_write = true;
+ config::enable_async_file_cache_write_inflight_write_buffer_index =
true;
+ config::enable_read_cache_file_directly = false;
+ config::enable_cache_read_from_peer = false;
+ config::clear_file_cache = true;
+ config::enable_evict_file_cache_in_advance = false;
+ config::file_cache_enter_disk_resource_limit_mode_percent = 99;
+ config::file_cache_each_block_size =
static_cast<int64_t>(FLAGS_block_size);
+ // Benchmark reads are external-table style with tablet_id=0, so they
do not register
+ // tablet-scoped TTL work. Short intervals only bound cache teardown
latency; production
+ // defaults let the otherwise idle TTL threads sleep for three minutes
before join().
+ config::file_cache_background_ttl_gc_interval_ms = 100;
+ config::file_cache_background_ttl_info_update_interval_ms = 100;
+ config::file_cache_background_tablet_id_flush_interval_ms = 100;
+ config::async_file_cache_write_workers_per_disk = FLAGS_reader_workers;
+ config::async_file_cache_write_max_pending_bytes_per_disk =
static_cast<int64_t>(
+ std::max(FLAGS_reader_operations, FLAGS_service_operations) *
FLAGS_block_size);
+
+ DORIS_CHECK(ExecEnv::GetInstance()->file_cache_factory() == nullptr);
+
ExecEnv::GetInstance()->set_file_cache_open_fd_cache(std::make_unique<FDCache>());
+ _factory = std::make_unique<FileCacheFactory>();
+ ExecEnv::GetInstance()->set_file_cache_factory(_factory.get());
+
+ FileCacheSettings settings;
+ settings.capacity = cache_capacity;
+ settings.max_file_block_size = FLAGS_block_size;
+ settings.max_query_cache_size = 0;
+ const size_t auxiliary_queue_size = FLAGS_block_size;
+ settings.disposable_queue_size = auxiliary_queue_size;
+ settings.disposable_queue_elements = 128;
+ settings.index_queue_size = auxiliary_queue_size;
+ settings.index_queue_elements = 128;
+ settings.ttl_queue_size = auxiliary_queue_size;
+ settings.ttl_queue_elements = 128;
+ settings.query_queue_size = cache_capacity - 3 * auxiliary_queue_size;
+ settings.query_queue_elements = std::max<size_t>(
+ 1024, 2 * std::max(FLAGS_reader_operations,
FLAGS_service_operations));
+
+ RETURN_IF_ERROR(_factory->create_file_cache(FLAGS_cache_path,
settings));
+ _cache = _factory->get_by_path(FLAGS_cache_path);
+ DORIS_CHECK(_cache != nullptr);
+ RETURN_IF_ERROR(wait_until([&]() { return
_cache->get_async_open_success(); },
+ "file cache initialization"));
+ return Status::OK();
+ }
+
+ /// Drain accepted writes, then synchronously invalidate and clear all
cached blocks.
+ Status clear_cache() {
+ RETURN_IF_ERROR(wait_for_idle());
+ std::string clear_result;
+ RETURN_IF_ERROR(_factory->clear_file_caches(true, &clear_result));
+ return wait_for_idle();
+ }
+
+ /// Apply one explicit worker/queue snapshot to the production service.
+ /// @param workers Active background writer count.
+ /// @param max_pending_bytes Maximum accepted buffer-capacity bytes,
including active workers.
+ Status configure_service(size_t workers, size_t max_pending_bytes) {
+ auto options = service()->options();
+ options.worker_count = workers;
+ options.max_pending_bytes = max_pending_bytes;
+ return service()->update_options(options);
+ }
+
+ /// Wait until both queue ownership and reader-visible inflight payloads
are gone.
+ Status wait_for_idle() {
+ return wait_until(
+ [&]() {
+ return service()->pending_count() == 0 &&
service()->queued_count() == 0 &&
+ index()->count() == 0;
+ },
+ "async write queue and inflight index to drain");
+ }
+
+ BlockFileCache* cache() const { return _cache; }
+ AsyncCacheWriteService* service() const { return
_cache->async_write_service(); }
+ InflightWriteBufferIndex* index() const { return
_cache->inflight_write_buffer_index(); }
+
+private:
+ std::unique_ptr<FileCacheFactory> _factory;
+ BlockFileCache* _cache {nullptr};
+ bool _owns_cache_path {false};
+};
+
+/// Verify that a complete range can be resolved from the final cache state.
+/// @param cache Target cache.
+/// @param hash Logical file hash.
+/// @param offset Range offset to verify.
+/// @param size Expected persisted bytes.
+Status verify_cached_range(BlockFileCache* cache, const UInt128Wrapper& hash,
size_t offset,
Review Comment:
[P2] Validate the persisted bytes, not only DOWNLOADED coverage. The direct
producers fill each task with a deterministic `operation % 251` pattern, but
the record drops that expectation and this helper never reads a block. The
reader case likewise validates the foreground remote bytes, then only checks
cache metadata after drain. Buffer-offset, append, or persisted-data corruption
would therefore still produce a successful benchmark sample despite the
README's validation guarantee. Preserve the expected pattern and read each
range back (or force a cache-only second read), rejecting byte mismatches.
##########
be/src/io/cache/benchmark/async_file_cache_write_microbench.cpp:
##########
@@ -0,0 +1,1099 @@
+// 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_common.h"
+
+#if defined(BE_TEST) && defined(BUILD_FILE_CACHE_MICROBENCH_TOOL)
+
+#include <gflags/gflags.h>
+#include <glog/logging.h>
+
+#include <algorithm>
+#include <atomic>
+#include <barrier>
+#include <chrono>
+#include <cmath>
+#include <cstddef>
+#include <cstdint>
+#include <cstdlib>
+#include <cstring>
+#include <filesystem>
+#include <iomanip>
+#include <iostream>
+#include <memory>
+#include <mutex>
+#include <numeric>
+#include <sstream>
+#include <string>
+#include <string_view>
+#include <thread>
+#include <utility>
+#include <vector>
+
+#include "cloud/config.h"
+#include "common/config.h"
+#include "common/status.h"
+#include "io/cache/async_cache_write_service.h"
+#include "io/cache/block_file_cache.h"
+#include "io/cache/block_file_cache_factory.h"
+#include "io/cache/cached_remote_file_reader.h"
+#include "io/cache/fs_file_cache_storage.h"
+#include "io/cache/inflight_write_buffer_index.h"
+#include "io/fs/file_reader.h"
+#include "io/fs/path.h"
+#include "runtime/exec_env.h"
+#include "runtime/thread_context.h"
+#include "util/cpu_info.h"
+#include "util/disk_info.h"
+#include "util/mem_info.h"
+#include "util/slice.h"
+#include "util/time.h"
+
+DEFINE_string(benchmark_mode, "all",
+ "Comma-separated benchmark groups: reader, service, index, or
all");
+DEFINE_string(cache_path, "./output/async_file_cache_write_microbench",
+ "Directory used by the real filesystem-backed BlockFileCache");
+DEFINE_uint64(block_size, 1024 * 1024,
+ "File-cache block size and stride used by cold reader misses");
+DEFINE_uint64(request_size, 64 * 1024, "Bytes returned to the caller by each
reader operation");
+DEFINE_uint64(reader_operations, 128, "Cold read operations in each sync/async
reader case");
+DEFINE_uint64(service_task_size, 1024 * 1024, "Payload bytes in each direct
service task");
+DEFINE_uint64(service_operations, 256, "Attempted tasks in each direct service
case");
+DEFINE_uint64(service_key_count, 64, "Logical remote files spread across
direct service tasks");
+DEFINE_uint64(index_operations_per_thread, 100000,
+ "Inflight-index lookups performed by each producer thread");
+DEFINE_uint64(index_key_count, 4096, "Keys used by the representative sharded
index case");
+DEFINE_int32(producer_threads, 16, "Concurrent foreground readers or task
producers");
+DEFINE_int32(reader_workers, 16, "Async write workers used by the reader
comparison");
+DEFINE_string(worker_counts, "1,4,16",
+ "Comma-separated async write worker counts used by service
scaling cases");
+DEFINE_uint64(repetitions, 5, "Measured repetitions of every selected
benchmark case");
+DEFINE_uint64(backpressure_pending_bytes, 64 * 1024 * 1024,
+ "Pending-buffer byte limit used by the saturated service case");
+DEFINE_uint64(queue_sample_interval_us, 50,
+ "Sampling interval for pending, queued, and inflight peak
values");
+DEFINE_uint64(timeout_seconds, 120, "Maximum time to wait for one benchmark
case to drain");
+DEFINE_bool(keep_cache, false, "Keep benchmark cache files after the process
exits");
+
+namespace doris::io {
+namespace {
+
+using Clock = std::chrono::steady_clock;
+using Nanoseconds = std::chrono::nanoseconds;
+
+constexpr size_t kMiB = 1024 * 1024;
+
+/// Compact percentile summary for foreground operation latency.
+struct LatencySummary {
+ double average_us {0};
+ double p50_us {0};
+ double p95_us {0};
+ double p99_us {0};
+ double maximum_us {0};
+};
+
+/// Return one nearest-rank percentile from an already sorted nanosecond
sample set.
+/// @param sorted_ns Ascending latency samples in nanoseconds.
+/// @param percentile Requested percentile in the inclusive range [0, 1].
+double percentile_us(const std::vector<int64_t>& sorted_ns, double percentile)
{
+ DORIS_CHECK(!sorted_ns.empty());
+ DORIS_CHECK(percentile > 0 && percentile <= 1);
+ const size_t index =
+ static_cast<size_t>(std::ceil(percentile *
static_cast<double>(sorted_ns.size()))) - 1;
+ return static_cast<double>(sorted_ns[std::min(index, sorted_ns.size() -
1)]) / 1000.0;
+}
+
+/// Merge per-thread samples and calculate stable latency percentiles.
+/// @param per_thread_ns Independently owned samples, one vector per producer.
+LatencySummary summarize_latencies(const std::vector<std::vector<int64_t>>&
per_thread_ns) {
+ size_t sample_count = 0;
+ for (const auto& samples : per_thread_ns) {
+ sample_count += samples.size();
+ }
+ DORIS_CHECK(sample_count > 0);
+
+ std::vector<int64_t> sorted_ns;
+ sorted_ns.reserve(sample_count);
+ for (const auto& samples : per_thread_ns) {
+ sorted_ns.insert(sorted_ns.end(), samples.begin(), samples.end());
+ }
+ std::sort(sorted_ns.begin(), sorted_ns.end());
+ const int64_t total_ns = std::accumulate(sorted_ns.begin(),
sorted_ns.end(), int64_t {0});
+ return LatencySummary {
+ .average_us =
+ static_cast<double>(total_ns) /
static_cast<double>(sample_count) / 1000.0,
+ .p50_us = percentile_us(sorted_ns, 0.50),
+ .p95_us = percentile_us(sorted_ns, 0.95),
+ .p99_us = percentile_us(sorted_ns, 0.99),
+ .maximum_us = static_cast<double>(sorted_ns.back()) / 1000.0,
+ };
+}
+
+/// Parse a comma-separated positive integer list used for worker scaling.
+/// @param text Raw gflag value such as "1,4,16".
+/// @param values Parsed worker counts in input order.
+Status parse_positive_integer_list(std::string_view text, std::vector<size_t>*
values) {
+ DORIS_CHECK(values != nullptr);
+ values->clear();
+ std::stringstream stream {std::string(text)};
+ std::string token;
+ while (std::getline(stream, token, ',')) {
+ try {
+ const long long value = std::stoll(token);
+ if (value <= 0) {
+ return Status::InvalidArgument("worker count must be positive:
{}", token);
+ }
+ values->push_back(static_cast<size_t>(value));
+ } catch (const std::exception& error) {
+ return Status::InvalidArgument("invalid worker count '{}': {}",
token, error.what());
+ }
+ }
+ if (values->empty()) {
+ return Status::InvalidArgument("worker_counts cannot be empty");
+ }
+ return Status::OK();
+}
+
+/// Split the selected benchmark groups while preserving a small command-line
surface.
+/// @param text Raw mode string.
+std::vector<std::string> parse_modes(std::string_view text) {
+ std::stringstream stream {std::string(text)};
+ std::vector<std::string> modes;
+ std::string mode;
+ while (std::getline(stream, mode, ',')) {
+ if (!mode.empty()) {
+ modes.emplace_back(std::move(mode));
+ }
+ }
+ return modes;
+}
+
+/// Return whether a benchmark group was requested explicitly or through "all".
+/// @param modes Parsed benchmark groups.
+/// @param target Group to test.
+bool mode_enabled(const std::vector<std::string>& modes, std::string_view
target) {
+ return std::find(modes.begin(), modes.end(), "all") != modes.end() ||
+ std::find(modes.begin(), modes.end(), target) != modes.end();
+}
+
+/// Validate sizes and concurrency before allocating cache capacity or
starting threads.
+Status validate_flags(const std::vector<std::string>& modes,
+ const std::vector<size_t>& worker_counts) {
+ if (modes.empty()) {
+ return Status::InvalidArgument("benchmark_mode cannot be empty");
+ }
+ for (const auto& mode : modes) {
+ if (mode != "all" && mode != "reader" && mode != "service" && mode !=
"index") {
+ return Status::InvalidArgument("unsupported benchmark mode: {}",
mode);
+ }
+ }
+ DORIS_CHECK(!worker_counts.empty());
+ if (FLAGS_producer_threads <= 0 || FLAGS_reader_workers <= 0) {
+ return Status::InvalidArgument("producer_threads and reader_workers
must be positive");
+ }
+ if (FLAGS_block_size == 0 || FLAGS_request_size == 0 || FLAGS_request_size
> FLAGS_block_size) {
+ return Status::InvalidArgument("sizes must satisfy 0 < request_size <=
block_size");
+ }
+ if (FLAGS_reader_operations <
static_cast<uint64_t>(FLAGS_producer_threads) ||
+ FLAGS_service_operations <
static_cast<uint64_t>(FLAGS_producer_threads)) {
+ return Status::InvalidArgument(
+ "reader_operations and service_operations must be at least
producer_threads");
+ }
+ if (FLAGS_service_task_size != FLAGS_block_size || FLAGS_service_key_count
== 0 ||
+ FLAGS_index_operations_per_thread == 0 || FLAGS_index_key_count == 0 ||
+ FLAGS_backpressure_pending_bytes == 0 ||
FLAGS_queue_sample_interval_us == 0 ||
+ FLAGS_timeout_seconds == 0 || FLAGS_repetitions == 0) {
+ return Status::InvalidArgument(
+ "operation counts, timeouts, and service_task_size ==
block_size are required");
+ }
+ return Status::OK();
+}
+
+/// Produce deterministic data without adding network or object-store latency
to reader results.
+class SyntheticRemoteFileReader final : public FileReader {
+public:
+ /// @param path Stable logical path used to derive the file-cache hash.
+ /// @param file_size Virtual file length; no payload is allocated for it.
+ /// @param block_size Pattern granularity used to validate copied bytes.
+ SyntheticRemoteFileReader(Path path, size_t file_size, size_t block_size)
+ : _path(std::move(path)), _file_size(file_size),
_block_size(block_size) {}
+
+ Status close() override {
+ _closed.store(true, std::memory_order_release);
+ return Status::OK();
+ }
+
+ const Path& path() const override { return _path; }
+ size_t size() const override { return _file_size; }
+ bool closed() const override { return
_closed.load(std::memory_order_acquire); }
+ int64_t mtime() const override { return 0; }
+
+ /// Return the byte value expected at an aligned benchmark block.
+ /// @param offset File offset inside the virtual source.
+ char expected_byte(size_t offset) const {
+ return static_cast<char>((offset / _block_size) % 251);
+ }
+
+protected:
+ /// Fill the requested span from a deterministic virtual file.
+ Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
+ const IOContext* io_ctx) override {
+ DORIS_CHECK(bytes_read != nullptr);
+ if (offset > _file_size || result.size > _file_size - offset) {
+ return Status::InvalidArgument("synthetic read [{}, {}) exceeds
file size {}", offset,
+ offset + result.size, _file_size);
+ }
+ size_t copied = 0;
+ while (copied < result.size) {
+ const size_t current_offset = offset + copied;
+ const size_t block_end =
+ std::min(_file_size, (current_offset / _block_size + 1) *
_block_size);
+ const size_t bytes = std::min(result.size - copied, block_end -
current_offset);
+ std::memset(result.data + copied, expected_byte(current_offset),
bytes);
+ copied += bytes;
+ }
+ *bytes_read = result.size;
+ return Status::OK();
+ }
+
+private:
+ Path _path;
+ size_t _file_size;
+ size_t _block_size;
+ std::atomic<bool> _closed {false};
+};
+
+/// Record the first worker-thread error without obscuring the performance hot
path.
+class ConcurrentError {
+public:
+ /// Store the first non-OK status observed by any producer.
+ void set(Status status) {
+ if (status.ok()) {
+ return;
+ }
+ bool expected = false;
+ if (_failed.compare_exchange_strong(expected, true,
std::memory_order_acq_rel)) {
+ std::lock_guard lock(_mutex);
+ _status = std::move(status);
+ }
+ }
+
+ bool failed() const { return _failed.load(std::memory_order_acquire); }
+
+ /// Return the stored error after producer threads have joined.
+ Status status() const {
+ std::lock_guard lock(_mutex);
+ return _status;
+ }
+
+private:
+ std::atomic<bool> _failed {false};
+ mutable std::mutex _mutex;
+ Status _status;
+};
+
+/// Sample public queue gauges so short benchmark cases still report a
meaningful high-water mark.
+class QueuePeakSampler {
+public:
+ /// @param service Service whose pending and queued gauges are sampled.
+ /// @param index Inflight index paired with the service.
+ QueuePeakSampler(AsyncCacheWriteService* service,
InflightWriteBufferIndex* index)
+ : _service(service), _index(index) {
+ DORIS_CHECK(_service != nullptr);
+ DORIS_CHECK(_index != nullptr);
+ }
+
+ /// Start the sampling thread. The sampler does not mutate benchmark state.
+ void start() {
+ _running.store(true, std::memory_order_release);
+ _thread = std::thread([this]() {
+ while (_running.load(std::memory_order_acquire)) {
+ sample();
+ std::this_thread::sleep_for(
+
std::chrono::microseconds(FLAGS_queue_sample_interval_us));
+ }
+ sample();
+ });
+ }
+
+ /// Stop sampling and join before reading peak values.
+ void stop() {
+ _running.store(false, std::memory_order_release);
+ if (_thread.joinable()) {
+ _thread.join();
+ }
+ }
+
+ ~QueuePeakSampler() { stop(); }
+
+ size_t peak_pending() const { return
_peak_pending.load(std::memory_order_relaxed); }
+ size_t peak_queued() const { return
_peak_queued.load(std::memory_order_relaxed); }
+ size_t peak_inflight() const { return
_peak_inflight.load(std::memory_order_relaxed); }
+ size_t peak_buffer_bytes() const { return
_peak_buffer_bytes.load(std::memory_order_relaxed); }
+
+private:
+ /// Capture one internally consistent set of independently sampled public
gauges.
+ void sample() {
+ update_max(&_peak_pending, _service->pending_count());
+ update_max(&_peak_queued, _service->queued_count());
+ update_max(&_peak_inflight, _index->count());
+ const int64_t buffer_memory_bytes = _service->buffer_memory_bytes();
+ DORIS_CHECK(buffer_memory_bytes >= 0);
+ update_max(&_peak_buffer_bytes,
static_cast<size_t>(buffer_memory_bytes));
+ }
+
+ /// Atomically preserve the largest sampled gauge value.
+ static void update_max(std::atomic<size_t>* maximum, size_t value) {
+ size_t current = maximum->load(std::memory_order_relaxed);
+ while (current < value &&
+ !maximum->compare_exchange_weak(current, value,
std::memory_order_relaxed)) {
+ }
+ }
+
+ AsyncCacheWriteService* _service;
+ InflightWriteBufferIndex* _index;
+ std::atomic<bool> _running {false};
+ std::thread _thread;
+ std::atomic<size_t> _peak_pending {0};
+ std::atomic<size_t> _peak_queued {0};
+ std::atomic<size_t> _peak_inflight {0};
+ std::atomic<size_t> _peak_buffer_bytes {0};
+};
+
+/// Poll one asynchronous completion predicate with a bounded failure mode.
+/// @param predicate Returns true after the expected state is reached.
+/// @param description Included in timeout diagnostics.
+template <typename Predicate>
+Status wait_until(Predicate&& predicate, std::string_view description) {
+ const auto deadline = Clock::now() +
std::chrono::seconds(FLAGS_timeout_seconds);
+ while (!predicate()) {
+ if (Clock::now() >= deadline) {
+ return Status::TimedOut("timed out waiting for {}", description);
+ }
+ std::this_thread::sleep_for(std::chrono::microseconds(100));
+ }
+ return Status::OK();
+}
+
+/// Own one real filesystem cache and install its factory into the otherwise
minimal ExecEnv.
+class BenchmarkEnvironment {
+public:
+ BenchmarkEnvironment() = default;
+
+ ~BenchmarkEnvironment() {
+ if (_factory) {
+ ExecEnv::GetInstance()->set_file_cache_factory(nullptr);
+ _factory.reset();
+ }
+ if (_owns_cache_path && !FLAGS_keep_cache) {
+ std::error_code error;
+ std::filesystem::remove_all(FLAGS_cache_path, error);
+ }
+ }
+
+ /// Configure globals, create the cache, and wait for its asynchronous
metadata open.
+ /// @param cache_capacity Bytes reserved for the benchmark's normal queue.
+ Status initialize(size_t cache_capacity) {
+ std::error_code error;
+ const auto cache_path =
+ std::filesystem::absolute(FLAGS_cache_path,
error).lexically_normal();
+ if (error) {
+ return Status::IOError("failed to resolve cache path {}: {}",
FLAGS_cache_path,
+ error.message());
+ }
+ const auto current_path = std::filesystem::current_path(error);
+ if (error) {
+ return Status::IOError("failed to resolve current directory: {}",
error.message());
+ }
+ if (cache_path == cache_path.root_path() || cache_path ==
current_path) {
+ return Status::InvalidArgument("cache path must be a dedicated
subdirectory: {}",
+ cache_path.string());
+ }
+ const bool cache_path_exists = std::filesystem::exists(cache_path,
error);
+ if (error) {
+ return Status::IOError("failed to inspect cache path {}: {}",
cache_path.string(),
+ error.message());
+ }
+ if (cache_path_exists) {
+ const bool cache_path_is_directory =
std::filesystem::is_directory(cache_path, error);
+ if (error) {
+ return Status::IOError("failed to inspect cache path {}: {}",
cache_path.string(),
+ error.message());
+ }
+ if (!cache_path_is_directory) {
+ return Status::InvalidArgument("cache path is not a directory:
{}",
+ cache_path.string());
+ }
+ const bool cache_path_is_empty =
std::filesystem::is_empty(cache_path, error);
+ if (error) {
+ return Status::IOError("failed to inspect cache path {}: {}",
cache_path.string(),
+ error.message());
+ }
+ if (!cache_path_is_empty) {
+ return Status::InvalidArgument(
+ "cache path must not exist or must be an empty
directory: {}",
+ cache_path.string());
+ }
+ } else {
+ std::filesystem::create_directories(cache_path, error);
+ if (error) {
+ return Status::IOError("failed to create cache path {}: {}",
cache_path.string(),
+ error.message());
+ }
+ }
+ _owns_cache_path = true;
+
+ config::enable_async_file_cache_write = true;
+ config::enable_async_file_cache_write_inflight_write_buffer_index =
true;
+ config::enable_read_cache_file_directly = false;
+ config::enable_cache_read_from_peer = false;
+ config::clear_file_cache = true;
+ config::enable_evict_file_cache_in_advance = false;
+ config::file_cache_enter_disk_resource_limit_mode_percent = 99;
+ config::file_cache_each_block_size =
static_cast<int64_t>(FLAGS_block_size);
+ // Benchmark reads are external-table style with tablet_id=0, so they
do not register
+ // tablet-scoped TTL work. Short intervals only bound cache teardown
latency; production
+ // defaults let the otherwise idle TTL threads sleep for three minutes
before join().
+ config::file_cache_background_ttl_gc_interval_ms = 100;
+ config::file_cache_background_ttl_info_update_interval_ms = 100;
+ config::file_cache_background_tablet_id_flush_interval_ms = 100;
+ config::async_file_cache_write_workers_per_disk = FLAGS_reader_workers;
+ config::async_file_cache_write_max_pending_bytes_per_disk =
static_cast<int64_t>(
+ std::max(FLAGS_reader_operations, FLAGS_service_operations) *
FLAGS_block_size);
+
+ DORIS_CHECK(ExecEnv::GetInstance()->file_cache_factory() == nullptr);
+
ExecEnv::GetInstance()->set_file_cache_open_fd_cache(std::make_unique<FDCache>());
+ _factory = std::make_unique<FileCacheFactory>();
+ ExecEnv::GetInstance()->set_file_cache_factory(_factory.get());
+
+ FileCacheSettings settings;
+ settings.capacity = cache_capacity;
+ settings.max_file_block_size = FLAGS_block_size;
+ settings.max_query_cache_size = 0;
+ const size_t auxiliary_queue_size = FLAGS_block_size;
+ settings.disposable_queue_size = auxiliary_queue_size;
+ settings.disposable_queue_elements = 128;
+ settings.index_queue_size = auxiliary_queue_size;
+ settings.index_queue_elements = 128;
+ settings.ttl_queue_size = auxiliary_queue_size;
+ settings.ttl_queue_elements = 128;
+ settings.query_queue_size = cache_capacity - 3 * auxiliary_queue_size;
+ settings.query_queue_elements = std::max<size_t>(
+ 1024, 2 * std::max(FLAGS_reader_operations,
FLAGS_service_operations));
+
+ RETURN_IF_ERROR(_factory->create_file_cache(FLAGS_cache_path,
settings));
+ _cache = _factory->get_by_path(FLAGS_cache_path);
+ DORIS_CHECK(_cache != nullptr);
+ RETURN_IF_ERROR(wait_until([&]() { return
_cache->get_async_open_success(); },
+ "file cache initialization"));
+ return Status::OK();
+ }
+
+ /// Drain accepted writes, then synchronously invalidate and clear all
cached blocks.
+ Status clear_cache() {
+ RETURN_IF_ERROR(wait_for_idle());
+ std::string clear_result;
+ RETURN_IF_ERROR(_factory->clear_file_caches(true, &clear_result));
+ return wait_for_idle();
+ }
+
+ /// Apply one explicit worker/queue snapshot to the production service.
+ /// @param workers Active background writer count.
+ /// @param max_pending_bytes Maximum accepted buffer-capacity bytes,
including active workers.
+ Status configure_service(size_t workers, size_t max_pending_bytes) {
+ auto options = service()->options();
+ options.worker_count = workers;
+ options.max_pending_bytes = max_pending_bytes;
+ return service()->update_options(options);
+ }
+
+ /// Wait until both queue ownership and reader-visible inflight payloads
are gone.
+ Status wait_for_idle() {
+ return wait_until(
+ [&]() {
+ return service()->pending_count() == 0 &&
service()->queued_count() == 0 &&
+ index()->count() == 0;
+ },
+ "async write queue and inflight index to drain");
+ }
+
+ BlockFileCache* cache() const { return _cache; }
+ AsyncCacheWriteService* service() const { return
_cache->async_write_service(); }
+ InflightWriteBufferIndex* index() const { return
_cache->inflight_write_buffer_index(); }
+
+private:
+ std::unique_ptr<FileCacheFactory> _factory;
+ BlockFileCache* _cache {nullptr};
+ bool _owns_cache_path {false};
+};
+
+/// Verify that a complete range can be resolved from the final cache state.
+/// @param cache Target cache.
+/// @param hash Logical file hash.
+/// @param offset Range offset to verify.
+/// @param size Expected persisted bytes.
+Status verify_cached_range(BlockFileCache* cache, const UInt128Wrapper& hash,
size_t offset,
+ size_t size) {
+ DORIS_CHECK(cache != nullptr);
+ ReadStatistics stats;
+ CacheContext context;
+ context.stats = &stats;
+ FileBlocks blocks;
+ bool fully_covered = false;
+ RETURN_IF_ERROR(cache->get_downloaded_blocks_if_fully_covered(hash,
offset, size, context,
+ &blocks,
&fully_covered));
+ if (!fully_covered) {
+ return Status::InternalError<false>("cache range [{}, {}) was not
persisted", offset,
+ offset + size);
+ }
+ return Status::OK();
+}
+
+/// Shared result fields printed for reader and direct-service cases.
+struct AsyncWriteResult {
+ std::string benchmark;
+ std::string variant;
+ size_t producers {0};
+ size_t workers {0};
+ size_t operations {0};
+ size_t accepted {0};
+ size_t rejected {0};
+ size_t evicted {0};
+ size_t persisted {0};
+ size_t bytes_per_operation {0};
+ double foreground_seconds {0};
+ double drain_seconds {0};
+ double total_seconds {0};
+ size_t peak_pending {0};
+ size_t peak_queued {0};
+ size_t peak_inflight {0};
+ size_t peak_buffer_bytes {0};
+ int64_t queue_lock_wait_p99_us {0};
+ int64_t queue_lock_hold_p99_us {0};
+ LatencySummary latency;
+};
+
+/// Print one machine-readable line without hiding the foreground/drain
distinction.
+/// @param result Completed benchmark result.
+/// @param repetition One-based repetition index.
+void print_async_write_result(const AsyncWriteResult& result, size_t
repetition) {
+ const double foreground_ops_per_sec =
+ static_cast<double>(result.operations) / result.foreground_seconds;
+ const double persisted_mib_per_sec =
+ result.total_seconds > 0
+ ? static_cast<double>(result.persisted *
result.bytes_per_operation) /
+ static_cast<double>(kMiB) / result.total_seconds
+ : 0;
+ std::cout << std::fixed << std::setprecision(3) << "RESULT"
+ << " benchmark=" << result.benchmark << " variant=" <<
result.variant
+ << " repetition=" << repetition << " producers=" <<
result.producers
+ << " workers=" << result.workers << " operations=" <<
result.operations
+ << " accepted=" << result.accepted << " rejected=" <<
result.rejected
+ << " evicted=" << result.evicted << " persisted=" <<
result.persisted
+ << " bytes_per_operation=" << result.bytes_per_operation
+ << " foreground_seconds=" << result.foreground_seconds
+ << " drain_seconds=" << result.drain_seconds
+ << " total_seconds=" << result.total_seconds
+ << " foreground_ops_per_sec=" << foreground_ops_per_sec
+ << " persisted_mib_per_sec=" << persisted_mib_per_sec
+ << " avg_us=" << result.latency.average_us << " p50_us=" <<
result.latency.p50_us
+ << " p95_us=" << result.latency.p95_us << " p99_us=" <<
result.latency.p99_us
+ << " max_us=" << result.latency.maximum_us << " peak_pending="
<< result.peak_pending
+ << " peak_queued=" << result.peak_queued << " peak_inflight=" <<
result.peak_inflight
+ << " peak_buffer_bytes=" << result.peak_buffer_bytes
+ << " queue_lock_wait_p99_us=" << result.queue_lock_wait_p99_us
+ << " queue_lock_hold_p99_us=" << result.queue_lock_hold_p99_us
<< '\n';
+}
+
+/// Compare cold-miss caller latency with synchronous and asynchronous cache
persistence.
+/// @param environment Shared real cache, cleared before the case.
+/// @param mode Explicit write policy applied to every CachedRemoteFileReader.
+/// @param variant Stable output label.
+/// @param repetition One-based repetition index included in output.
+Status run_reader_case(BenchmarkEnvironment* environment, CacheWriteMode mode,
std::string variant,
+ size_t repetition) {
+ DORIS_CHECK(environment != nullptr);
+ RETURN_IF_ERROR(environment->clear_cache());
+ RETURN_IF_ERROR(environment->configure_service(
+ static_cast<size_t>(FLAGS_reader_workers),
+ static_cast<size_t>(FLAGS_reader_operations +
FLAGS_reader_workers) *
+ FLAGS_block_size));
+
+ const size_t producer_count = static_cast<size_t>(FLAGS_producer_threads);
+ const size_t operation_count =
static_cast<size_t>(FLAGS_reader_operations);
+ const size_t file_size = operation_count * FLAGS_block_size;
+ const std::string file_name = "async_write_reader_" + variant + ".bin";
+ const Path path("/synthetic/" + file_name);
+
+ std::vector<std::thread> producers;
+ producers.reserve(producer_count);
+ std::vector<std::vector<int64_t>> latencies(producer_count);
+ std::vector<FileCacheStatistics> statistics(producer_count);
+ std::barrier start_barrier(static_cast<std::ptrdiff_t>(producer_count +
1));
+ ConcurrentError error;
+ QueuePeakSampler sampler(environment->service(), environment->index());
+ sampler.start();
+
+ for (size_t producer = 0; producer < producer_count; ++producer) {
+ producers.emplace_back([&, producer]() {
+ SCOPED_INIT_THREAD_CONTEXT();
+ auto remote =
+ std::make_shared<SyntheticRemoteFileReader>(path,
file_size, FLAGS_block_size);
+ FileReaderOptions options;
+ options.cache_type = FileCachePolicy::FILE_BLOCK_CACHE;
+ options.cache_write_mode = mode;
+ options.cache_base_path = FLAGS_cache_path;
+ options.tablet_id = 0;
+ auto reader = std::make_shared<CachedRemoteFileReader>(remote,
options);
+ IOContext io_context;
+ io_context.file_cache_stats = &statistics[producer];
+ io_context.bypass_peer_read = true;
+ std::string buffer(FLAGS_request_size, '\0');
+ latencies[producer].reserve((operation_count + producer_count - 1)
/ producer_count);
+
+ start_barrier.arrive_and_wait();
+ for (size_t operation = producer; operation < operation_count;
+ operation += producer_count) {
+ if (error.failed()) {
+ break;
+ }
+ const size_t offset = operation * FLAGS_block_size;
+ size_t bytes_read = 0;
+ const auto start = Clock::now();
+ Status status = reader->read_at(offset, Slice(buffer.data(),
buffer.size()),
+ &bytes_read, &io_context);
+ const auto elapsed =
std::chrono::duration_cast<Nanoseconds>(Clock::now() - start);
+ latencies[producer].push_back(elapsed.count());
+ if (!status.ok()) {
+ error.set(std::move(status));
+ break;
+ }
+ if (bytes_read != buffer.size() ||
+ buffer.front() != remote->expected_byte(offset) ||
+ buffer.back() != remote->expected_byte(offset +
buffer.size() - 1)) {
+ error.set(Status::InternalError(
+ "reader result mismatch at operation {},
bytes_read={}", operation,
+ bytes_read));
+ break;
+ }
+ }
+ });
+ }
+
+ const auto foreground_start = Clock::now();
+ start_barrier.arrive_and_wait();
+ for (auto& producer : producers) {
+ producer.join();
+ }
+ const auto foreground_end = Clock::now();
+ if (error.failed()) {
+ sampler.stop();
+ return error.status();
+ }
+
+ RETURN_IF_ERROR(environment->wait_for_idle());
+ const auto drain_end = Clock::now();
+ sampler.stop();
+
+ FileCacheStatistics total_stats;
+ for (const auto& local_stats : statistics) {
+ total_stats.merge_from(local_stats);
+ }
+ const auto hash = BlockFileCache::hash(path.native() + ":0");
+ RETURN_IF_ERROR(verify_cached_range(environment->cache(), hash, 0,
file_size));
+
+ const double foreground_seconds =
+ std::chrono::duration<double>(foreground_end -
foreground_start).count();
+ const double drain_seconds = std::chrono::duration<double>(drain_end -
foreground_end).count();
+ AsyncWriteResult result {
+ .benchmark = "reader",
+ .variant = std::move(variant),
+ .producers = producer_count,
+ .workers = mode == CacheWriteMode::ASYNC_WRITE
+ ? static_cast<size_t>(FLAGS_reader_workers)
+ : 0,
+ .operations = operation_count,
+ .accepted = mode == CacheWriteMode::ASYNC_WRITE
+ ?
static_cast<size_t>(total_stats.async_cache_write_submitted)
+ : 0,
+ .rejected =
static_cast<size_t>(total_stats.async_cache_write_rejected),
+ .persisted = operation_count,
+ .bytes_per_operation = FLAGS_block_size,
+ .foreground_seconds = foreground_seconds,
+ .drain_seconds = drain_seconds,
+ .total_seconds = foreground_seconds + drain_seconds,
+ .peak_pending = sampler.peak_pending(),
+ .peak_queued = sampler.peak_queued(),
+ .peak_inflight = sampler.peak_inflight(),
+ .peak_buffer_bytes = sampler.peak_buffer_bytes(),
+ .latency = summarize_latencies(latencies),
+ };
+ print_async_write_result(result, repetition);
+ return Status::OK();
+}
+
+/// One accepted direct-service task retained for post-timing persistence
verification.
+struct ServiceTaskRecord {
+ UInt128Wrapper hash;
+ size_t offset {0};
+ size_t size {0};
+};
+
+/// Exercise producer admission, inflight publication, locked FIFO
consumption, and persistence.
+/// @param environment Shared real cache, cleared before the case.
+/// @param variant Stable output label.
+/// @param workers Active service consumers.
+/// @param max_pending_bytes Bounded pending-buffer byte limit.
+/// @param repetition One-based repetition index included in output.
+Status run_service_case(BenchmarkEnvironment* environment, std::string
variant, size_t workers,
+ size_t max_pending_bytes, size_t repetition) {
+ DORIS_CHECK(environment != nullptr);
+ RETURN_IF_ERROR(environment->clear_cache());
+ RETURN_IF_ERROR(environment->configure_service(workers,
max_pending_bytes));
+
+ const size_t producer_count = static_cast<size_t>(FLAGS_producer_threads);
+ const size_t operation_count =
static_cast<size_t>(FLAGS_service_operations);
+ std::vector<std::thread> producers;
+ producers.reserve(producer_count);
+ std::vector<std::vector<int64_t>> latencies(producer_count);
+ std::vector<std::vector<ServiceTaskRecord>>
accepted_records(producer_count);
+ std::barrier start_barrier(static_cast<std::ptrdiff_t>(producer_count +
1));
+ std::atomic<size_t> accepted {0};
+ std::atomic<size_t> rejected {0};
+ std::atomic<size_t> completed {0};
+ ConcurrentError error;
+ QueuePeakSampler sampler(environment->service(), environment->index());
+ const uint64_t baseline_evicted =
environment->service()->evicted_oldest_count();
+ sampler.start();
+
+ for (size_t producer = 0; producer < producer_count; ++producer) {
+ producers.emplace_back([&, producer]() {
+ SCOPED_INIT_THREAD_CONTEXT();
+ latencies[producer].reserve((operation_count + producer_count - 1)
/ producer_count);
+ accepted_records[producer].reserve((operation_count +
producer_count - 1) /
+ producer_count);
+ start_barrier.arrive_and_wait();
+
+ for (size_t operation = producer; operation < operation_count;
+ operation += producer_count) {
+ if (error.failed()) {
+ break;
+ }
+ const size_t key_index = operation % FLAGS_service_key_count;
+ const size_t block_index = operation / FLAGS_service_key_count;
+ const auto hash = BlockFileCache::hash("async_write_service_"
+ variant + "_" +
+
std::to_string(key_index));
+ const size_t offset = block_index * FLAGS_block_size;
+ const auto start = Clock::now();
+
+ AsyncCacheWriteBufferPtr buffer;
+ Status status =
environment->service()->allocate_tracked_buffer(
+ FLAGS_service_task_size, &buffer);
+ if (!status.ok()) {
+ error.set(std::move(status));
+ break;
+ }
+ std::memset(buffer->data(), static_cast<int>(operation % 251),
buffer->size());
+ const uint64_t epoch =
environment->service()->current_write_epoch();
+ auto entry = std::make_shared<InflightWriteBufferEntry>(
+ buffer, offset, buffer->size(), MonotonicMicros(),
epoch);
+ auto existing = environment->index()->insert_if_absent(hash,
offset, entry);
+ if (existing != nullptr) {
+ error.set(Status::InternalError(
+ "unexpected duplicate inflight owner at operation
{}", operation));
+ break;
+ }
+
+ AsyncCacheWriteTask task {
+ .cache_hash = hash,
+ .file_offset = offset,
+ .write_size = buffer->size(),
+ .buffer = buffer,
+ .admission_ctx = {},
+ .submit_ts_us = MonotonicMicros(),
+ .write_epoch = epoch,
+ .on_finalized =
+ [index = environment->index(), hash, offset,
entry,
+ &completed](const AsyncCacheWriteTask&) {
+ index->remove_if(hash, offset, entry);
+ completed.fetch_add(1,
std::memory_order_release);
+ },
+ };
+ const bool submitted =
environment->service()->try_submit(std::move(task));
+ const auto elapsed =
std::chrono::duration_cast<Nanoseconds>(Clock::now() - start);
+ latencies[producer].push_back(elapsed.count());
+ if (submitted) {
+ accepted.fetch_add(1, std::memory_order_relaxed);
+ accepted_records[producer].push_back(ServiceTaskRecord {
+ .hash = hash, .offset = offset, .size =
FLAGS_service_task_size});
+ } else {
+ environment->index()->remove_if(hash, offset, entry);
+ environment->index()->record_backpressure_rollback();
+ rejected.fetch_add(1, std::memory_order_relaxed);
+ }
+ }
+ });
+ }
+
+ const auto foreground_start = Clock::now();
+ start_barrier.arrive_and_wait();
+ for (auto& producer : producers) {
+ producer.join();
+ }
+ const auto foreground_end = Clock::now();
+ if (error.failed()) {
+ sampler.stop();
+ return error.status();
+ }
+
+ RETURN_IF_ERROR(wait_until(
+ [&]() {
+ return environment->service()->pending_count() == 0 &&
+ completed.load(std::memory_order_acquire) ==
+ accepted.load(std::memory_order_acquire) &&
+ environment->index()->count() == 0;
+ },
+ "direct service tasks to complete"));
+ const auto drain_end = Clock::now();
+ sampler.stop();
+
+ size_t persisted = 0;
+ for (const auto& records : accepted_records) {
+ for (const auto& record : records) {
+ Status status = verify_cached_range(environment->cache(),
record.hash, record.offset,
+ record.size);
+ if (status.ok()) {
+ ++persisted;
+ }
+ }
+ }
+ const size_t accepted_count = accepted.load(std::memory_order_relaxed);
+ DORIS_CHECK(persisted <= accepted_count);
+ const size_t evicted =
+ static_cast<size_t>(environment->service()->evicted_oldest_count()
- baseline_evicted);
+ if (accepted_count - persisted != evicted) {
+ return Status::InternalError("{} accepted tasks produced {} persisted
and {} evicted",
+ accepted_count, persisted, evicted);
+ }
+
+ const double foreground_seconds =
+ std::chrono::duration<double>(foreground_end -
foreground_start).count();
+ const double drain_seconds = std::chrono::duration<double>(drain_end -
foreground_end).count();
+ AsyncWriteResult result {
+ .benchmark = "service",
+ .variant = std::move(variant),
+ .producers = producer_count,
+ .workers = workers,
+ .operations = operation_count,
+ .accepted = accepted_count,
+ .rejected = rejected.load(std::memory_order_relaxed),
+ .evicted = evicted,
+ .persisted = persisted,
+ .bytes_per_operation = FLAGS_service_task_size,
+ .foreground_seconds = foreground_seconds,
+ .drain_seconds = drain_seconds,
+ .total_seconds = foreground_seconds + drain_seconds,
+ .peak_pending = sampler.peak_pending(),
+ .peak_queued = sampler.peak_queued(),
+ .peak_inflight = sampler.peak_inflight(),
+ .peak_buffer_bytes = sampler.peak_buffer_bytes(),
+ .queue_lock_wait_p99_us =
environment->service()->queue_lock_wait_p99_us(),
Review Comment:
[P2] Report a case-local lock percentile here. `run_benchmarks()` reuses one
service across the reader workload, every worker-count/saturation variant, and
all repetitions, while these `bvar::LatencyRecorder`s are never reset or
baselined. Each RESULT therefore labels a rolling mixture of earlier cases as
the current variant's P99 (and reader cases print zero), so the advertised
worker-count/contention comparison is not interpretable. Recreate/reset the
recorder at an idle case boundary or collect case-owned lock samples.
--
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]
