github-actions[bot] commented on code in PR #67292: URL: https://github.com/apache/doris/pull/67292#discussion_r4413394781
########## be/src/io/cache/partial_block_writeback_manager.cpp: ########## @@ -0,0 +1,1035 @@ +// 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/partial_block_writeback_manager.h" + +#include <algorithm> +#include <chrono> +#include <cstring> +#include <iterator> +#include <optional> +#include <utility> +#include <vector> + +#include "common/logging.h" +#include "cpp/sync_point.h" +#include "io/cache/hole_fill_planner.h" +#include "io/cache/inflight_write_buffer_index.h" +#include "io/fs/read_ahead_metrics.h" +#include "io/fs/read_io_trace.h" +#include "util/countdown_latch.h" +#include "util/defer_op.h" +#include "util/thread.h" +#include "util/threadpool.h" +#include "util/time.h" + +namespace doris::io { +namespace { + +using namespace std::chrono_literals; + +constexpr auto kCapacityRetryInterval = 10ms; +constexpr size_t kMaxHoleFillWorkerCount = 128; + +} // namespace + +size_t PartialBlockWritebackManager::BlockKeyHash::operator()(const BlockKey& key) const { + const size_t manager_hash = std::hash<AsyncCacheWriteManager*> {}(key.write_manager); + const size_t file_hash = KeyHash {}(key.cache_hash); + const size_t offset_hash = std::hash<size_t> {}(key.block_offset); + return manager_hash ^ (file_hash << 1) ^ (offset_hash << 2); +} + +// Per-call diagnostics only. Keep the request identity alive across moves into a Task and emit +// once, after every manager/fragment lock has left scope. The disabled path reads no clocks. +struct PartialBlockWritebackManager::SubmitTrace : HoleSubmitTiming { + explicit SubmitTrace(const PartialBlockWritebackRequest& request) { + if (ReadIOTrace::enabled()) { + start_ns = MonotonicNanos(); + context = request.io_context; + reader = request.source_reader; + offset = request.block_offset + request.fragment_offset; + size = request.data.size; + } + } + + ~SubmitTrace() { + if (start_ns != 0) { + ReadIOTrace::record({.event = "hole_submit", + .context = &context.io_context, + .file = reader->path().native(), + .id = task_id, + .parent_id = context.io_context.read_trace_id, + .offset = offset, + .size = size, + .time_ns = MonotonicNanos(), + .start_ns = start_ns, + .outcome = outcome, + .hole_submit_timing = this}); + } + } + + int64_t now() const { return start_ns != 0 ? MonotonicNanos() : 0; } + + PartialBlockSubmitResult finish(PartialBlockSubmitResult result) { + switch (result) { + case PartialBlockSubmitResult::QUEUED: + outcome = "queued"; + break; + case PartialBlockSubmitResult::MERGED: + outcome = "merged"; + break; + case PartialBlockSubmitResult::ACTIVE_DEDUPLICATED: + outcome = "active_deduplicated"; + break; + case PartialBlockSubmitResult::CACHE_BLOCK_PRESENT: + outcome = "cache_block_present"; + break; + case PartialBlockSubmitResult::STALE_EPOCH: + outcome = "stale_epoch"; + break; + case PartialBlockSubmitResult::BUFFER_ALLOCATION_FAILED: + outcome = "allocation_failed"; + break; + case PartialBlockSubmitResult::REJECTED: + outcome = "rejected"; + break; + } + return result; + } + + int64_t start_ns {0}; + FileRangeReadIOContext context; + FileReaderSPtr reader; + size_t offset {0}; + size_t size {0}; + uint64_t task_id {0}; + std::string_view outcome; +}; + +// One record per locked scan, not per entry. No JSON formatting or trace-writer locking is done +// under the queue mutex. CV sleep and its mutex reacquisition are excluded from scan time. +struct PartialBlockWritebackManager::QueueScanTrace : HoleQueueScanStats { + int64_t start_ns {0}; + int64_t end_ns {0}; + + void record(bool activated) const { + if (start_ns != 0) { + ReadIOTrace::record({.event = "hole_queue_scan", + .time_ns = end_ns, + .start_ns = start_ns, + .outcome = activated ? "activated" : "no_runnable_task", + .queue_scan = this}); + } + } +}; + +struct PartialBlockWritebackManager::Task { + // Allocate one tracked block buffer and copy the first fragment into it. + static TaskPtr create(PartialBlockWritebackRequest request, const BlockKey& task_key, + size_t block_size, SubmitTrace& trace); + + // A queued task may merge while the manager lock is free. Once activate() wins, later + // fragments are deduplicated and the worker obtains their bytes from the source read instead. + std::optional<PartialBlockSubmitResult> try_merge(size_t fragment_offset, Slice data, + const IOContext& fragment_context, + SubmitTrace& trace) { + const auto lock_start = trace.now(); + std::lock_guard lock(fragment_mutex); + const auto locked_at = trace.now(); + trace.fragment_lock_wait_ns += locked_at - lock_start; + Defer hold_time {[&] { trace.fragment_lock_hold_ns += trace.now() - locked_at; }}; + trace.task_id = io_context.io_context.read_trace_id; + if (is_active()) { + const auto trace_start = trace.now(); + trace_fragment("fragment_ignored_active", fragment_offset, data.size, fragment_context); + trace.lifecycle_trace_ns += trace.now() - trace_start; + return PartialBlockSubmitResult::ACTIVE_DEDUPLICATED; + } + if (!key.write_manager->is_current_write_epoch(write_epoch)) { + return std::nullopt; + } + + TEST_SYNC_POINT("PartialBlockWritebackManager::try_submit:before_merge_copy"); + const auto copy_start = trace.now(); + std::memcpy(buffer->data() + fragment_offset, data.data, data.size); + trace.copy_ns += trace.now() - copy_start; + trace.copied_bytes += data.size; + covered_intervals.emplace_back( + FileRange {.offset = key.block_offset + fragment_offset, .size = data.size}); + const auto trace_start = trace.now(); + trace_fragment("fragment_merged", fragment_offset, data.size, fragment_context); + trace.lifecycle_trace_ns += trace.now() - trace_start; + return PartialBlockSubmitResult::MERGED; + } + + void trace_fragment(std::string_view event, size_t offset, size_t size, + const IOContext& fragment_context) const { + if (ReadIOTrace::enabled()) { + ReadIOTrace::record({.event = event, + .context = &fragment_context, + .file = source_reader->path().native(), + .id = io_context.io_context.read_trace_id, + .parent_id = fragment_context.read_trace_id, + .offset = key.block_offset + offset, + .size = size}); + } + } + + // Activation closes the merge window. Taking fragment_mutex here also includes a merge that + // entered just before activation, so covered_intervals is stable for the hole complement. + Status plan_hole_reads(const FileRangeCoalesceOptions& options, + std::vector<FileRange>* read_ranges) { + DORIS_CHECK(is_active()); + std::lock_guard lock(fragment_mutex); + const FileRange block {.offset = key.block_offset, .size = block_valid_size}; + return HoleFillPlanner::plan(block, covered_intervals, options, read_ranges); + } + + // Read a single range on the block worker; submit multiple ranges through its reusable token + // and join them, including on failure. + Status read_holes(const std::vector<FileRange>& read_ranges, ThreadPoolToken& token); + + void activate() { + activation_trace_ns = ReadIOTrace::enabled() ? MonotonicNanos() : 0; + bool expected = false; + DORIS_CHECK(active.compare_exchange_strong(expected, true, std::memory_order_acq_rel)); + } + + bool is_active() const { return active.load(std::memory_order_acquire); } + + // Only inspect epoch and inflight state under the queue mutex. Full cache probing follows + // task activation, outside that mutex. + bool should_discard_before_read() const { + if (!key.write_manager->accepting() || + !key.write_manager->is_current_write_epoch(write_epoch)) { + return true; + } + return inflight_index != nullptr && + inflight_index->lookup(key.cache_hash, key.block_offset) != nullptr; + } + + BlockKey key; + InflightWriteBufferIndex* inflight_index {nullptr}; + FileReaderSPtr source_reader; + size_t block_valid_size {0}; + AsyncCacheWriteBufferPtr buffer; + std::vector<FileRange> covered_intervals; + CacheAdmissionContext admission_ctx; + AsyncCacheWriteEpoch write_epoch; + FileRangeReadIOContext io_context; + // Set at successful queue admission under the manager mutex; merges leave it unchanged. + std::chrono::steady_clock::time_point enqueued_at; + int64_t activation_trace_ns {0}; // only read by the worker after taking the task + // Valid only while this task is in the manager queue; accessed under the manager mutex. + Queue::iterator queue_position; + // A queued task may absorb foreground fragments while unrelated queue operations proceed. + std::mutex fragment_mutex; + std::atomic<bool> active {false}; + +private: + // Read one buffer slice with per-call IO statistics on a block worker or remote-read thread. + Status _read_hole(const FileRange& range); +}; + +class PartialBlockWritebackManager::Worker : public std::enable_shared_from_this<Worker> { +public: + explicit Worker(PartialBlockWritebackManager& manager) : _manager(manager) {} + + Status start() { + auto self = shared_from_this(); + return _manager._worker_pool->submit_func([self = std::move(self)]() { self->_run(); }); + } + + // The caller holds the manager mutex so the changed wait predicate and notification are + // published atomically to an idle worker. + void request_stop() { _stop_requested.store(true, std::memory_order_release); } + + bool stop_requested() const { return _stop_requested.load(std::memory_order_acquire); } + + void wait_until_stopped() { _stopped.wait(); } + +private: + void _run() { + _manager._running_worker_count.fetch_add(1, std::memory_order_relaxed); + Defer mark_stopped {[this]() { + const size_t old_running = + _manager._running_worker_count.fetch_sub(1, std::memory_order_relaxed); + DCHECK_GT(old_running, 0); + _stopped.count_down(); + }}; + + // Destroy the token before mark_stopped lets shutdown release the remote-read pool. + auto remote_read_token = + _manager._remote_read_pool->new_token(ThreadPool::ExecutionMode::CONCURRENT); + while (!stop_requested()) { + auto task = _manager._take_ready_task(*this); + if (task == nullptr) { + return; + } + _manager._process_task(task, *remote_read_token); + } + } + + PartialBlockWritebackManager& _manager; + std::atomic<bool> _stop_requested {false}; + CountDownLatch _stopped {1}; +}; + +Status PartialBlockWritebackOptions::validate() const { + RETURN_IF_ERROR(hole_fill_coalesce.validate()); + if (block_size == 0) { + return Status::InvalidArgument("partial block writeback block size must be positive"); + } + if (worker_count == 0 || worker_count > kMaxHoleFillWorkerCount) { + return Status::InvalidArgument("partial block writeback worker count {} is invalid", + worker_count); + } + if (remote_read_thread_count <= 0) { + return Status::InvalidArgument("partial block remote-read thread count must be positive"); + } + if (merge_delay_ms < 0) { + return Status::InvalidArgument("partial block merge delay must be nonnegative"); + } + if (max_pending_bytes < block_size) { + return Status::InvalidArgument( + "partial block writeback pending bytes {} must hold at least one block of {} bytes", + max_pending_bytes, block_size); + } + return Status::OK(); +} + +PartialBlockWritebackManager::PartialBlockWritebackManager(PartialBlockWritebackOptions options) + : _options(std::move(options)), + _max_pending_tasks(_options.max_pending_bytes / _options.block_size), + _merge_delay(_options.merge_delay_ms), + _configured_worker_count(_options.worker_count) { + DORIS_CHECK(_max_pending_tasks > 0); + read_ahead_bvars().hole_fill_pending_bytes << 0; +} + +PartialBlockWritebackManager::~PartialBlockWritebackManager() { + shutdown(); +} + +Status PartialBlockWritebackManager::create( + const PartialBlockWritebackOptions& options, + std::unique_ptr<PartialBlockWritebackManager>* output_manager) { + DORIS_CHECK(output_manager != nullptr); + RETURN_IF_ERROR(options.validate()); + auto manager = std::unique_ptr<PartialBlockWritebackManager>( + new PartialBlockWritebackManager(options)); + RETURN_IF_ERROR(manager->_start()); + *output_manager = std::move(manager); + return Status::OK(); +} + +Status PartialBlockWritebackManager::_start() { + std::lock_guard lifecycle_lock(_lifecycle_mutex); + DORIS_CHECK(_worker_pool == nullptr); + const size_t worker_count = _configured_worker_count.load(std::memory_order_acquire); + RETURN_IF_ERROR(ThreadPoolBuilder("HoleFillWorkerPool") + .set_min_threads(static_cast<int>(worker_count)) + .set_max_threads(static_cast<int>(worker_count)) + .set_max_queue_size(static_cast<int>(kMaxHoleFillWorkerCount)) + .build(&_worker_pool)); + RETURN_IF_ERROR(ThreadPoolBuilder("HoleFillRemoteReadPool") + // Keep a reader available even if creating additional threads fails. + .set_min_threads(1) + .set_max_threads(_options.remote_read_thread_count) + .build(&_remote_read_pool)); + { + std::lock_guard lock(_mutex); + DORIS_CHECK(!_accepting); + _accepting = true; + } + RETURN_IF_ERROR(_resize_workers_locked(worker_count)); + return Thread::create( + "hole fill", "HoleFillDispatch", [this] { _dispatch_loop(); }, &_dispatch_thread); +} + +PartialBlockSubmitResult PartialBlockWritebackManager::try_submit( + PartialBlockWritebackRequest request) { + SubmitTrace trace(request); + request.sanity_check(_options.block_size); + + if (!request.write_manager->accepting()) { + return trace.finish(PartialBlockSubmitResult::REJECTED); + } + if (!request.write_manager->check_write_epoch(request.write_epoch)) { + return trace.finish(PartialBlockSubmitResult::STALE_EPOCH); + } + const auto probe_start = trace.now(); + const bool skip = request.write_manager->should_skip_block_writeback( + request.cache_hash, request.block_offset, request.block_valid_size, + request.admission_ctx, request.inflight_index); + trace.cache_probe_ns += trace.now() - probe_start; + if (skip) { + return trace.finish(PartialBlockSubmitResult::CACHE_BLOCK_PRESENT); + } + + const BlockKey key {.write_manager = request.write_manager, + .cache_hash = request.cache_hash, + .block_offset = request.block_offset}; + const size_t fragment_offset = request.fragment_offset; + const Slice fragment = request.data; + // Retain the submitting query/range identity even when a different query owns the task. + const IOContext fragment_context = request.io_context.io_context; + + TaskPtr existing; + { + const auto lock_start = trace.now(); + TEST_SYNC_POINT("PartialBlockWritebackManager::try_submit:before_queue_lock"); + std::lock_guard lock(_mutex); + const auto locked_at = trace.now(); + trace.queue_lock_wait_ns += locked_at - lock_start; + Defer hold_time {[&] { trace.queue_lock_hold_ns += trace.now() - locked_at; }}; + trace.queue_size = _queue.size(); + if (!_accepting) { + return trace.finish(PartialBlockSubmitResult::REJECTED); + } + const auto entry = _tasks.find(key); + if (entry != _tasks.end()) { + existing = entry->second; + } else if (_tasks.size() == _max_pending_tasks && _queue.empty()) { + return trace.finish(PartialBlockSubmitResult::REJECTED); + } + } + if (existing != nullptr) { Review Comment: [P2] Do not merge fragments into an evicted partial-block task. `try_submit()` finds an existing queued task under `_mutex`, releases that lock, and then calls `Task::try_merge()` at this line. While the lock is released, another submission can fill `_max_pending_tasks` and evict this inactive task from both `_tasks` and `_queue`. The shared pointer keeps the orphan alive, so its still-current epoch passes and `try_merge()` copies the fragment and returns `MERGED`, but no worker can dispatch that task; the caller reports success while the consumed bytes are lost. The same window exists after the `EXISTING` merge at lines 448-455. Revalidate that the task is still the current queued owner under the manager lock, or retry enqueueing the candidate before accepting the merge. ########## be/src/storage/segment/page_io.cpp: ########## @@ -36,11 +36,13 @@ #include "io/cache/cached_remote_file_reader.h" #include "io/fs/file_reader.h" #include "io/fs/file_writer.h" +#include "io/fs/read_ahead_metrics.h" #include "runtime/runtime_profile.h" #include "storage/cache/page_cache.h" #include "storage/olap_common.h" #include "storage/segment/encoding_info.h" Review Comment: [P2] Count only actual read-ahead fallbacks. `read_ahead_stats` is shared by the whole segment, but non-scalar/nested iterators are deliberately left on their original `FileReader`. When such a DATA_PAGE reaches this branch, the dynamic cast at line 178 returns null, yet the code still increments `fallback_pages`, `fallback_bytes`, and `fallback_time` before the ordinary read. `ReadAheadFallback*` profile counters therefore report nested reads as failed or rejected read-ahead fallbacks. Gate this accounting on a successful `SegmentReadAheadFileReader` attempt and add nested-column coverage. -- 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]
