github-actions[bot] commented on code in PR #67292: URL: https://github.com/apache/doris/pull/67292#discussion_r4358330833
########## be/src/io/fs/read_io_trace.cpp: ########## @@ -0,0 +1,317 @@ +// 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/fs/read_io_trace.h" + +#include <bvar/reducer.h> +#include <rapidjson/stringbuffer.h> +#include <rapidjson/writer.h> +#include <unistd.h> + +#include <atomic> +#include <string> +#include <utility> + +#include "common/cast_set.h" +#include "common/config.h" +#include "common/logging.h" +#include "cpp/sync_point.h" +#include "io/fs/file_writer.h" +#include "io/fs/local_file_system.h" +#include "io/io_common.h" +#include "runtime/thread_context.h" +#include "util/thread.h" +#include "util/time.h" +#include "util/uid_util.h" + +namespace doris::io { +namespace { + +std::atomic<uint64_t> read_trace_next_id {0}; +std::atomic<uint64_t> read_trace_event_count {0}; +bvar::Adder<uint64_t> read_trace_logged_events {"doris_read_io_trace_events"}; +bvar::Adder<uint64_t> read_trace_dropped_events {"doris_read_io_trace_dropped_events"}; +bvar::Adder<int64_t> read_trace_pending_bytes {"doris_read_io_trace_pending_bytes"}; +bvar::Adder<int64_t> read_trace_record_time_ns {"doris_read_io_trace_record_time_ns"}; + +const std::string& read_trace_process() { + // Distinguish BE restarts even when the operating system reuses a PID. + static const std::string process = + std::to_string(getpid()) + "-" + std::to_string(UnixMicros()); + return process; +} + +ReadIOTraceWriter& read_trace_writer() { + static ReadIOTraceWriter writer( + config::read_io_trace_dir.empty() ? (Path(FLAGS_log_dir) / "read_io_trace").native() + : config::read_io_trace_dir, + cast_set<size_t>(config::read_io_trace_flush_bytes), + std::chrono::milliseconds(config::read_io_trace_flush_interval_ms)); + return writer; +} + +const char* read_trace_source(const IOContext* context) { + if (context == nullptr) { + return "other"; + } + switch (context->read_trace_source) { + case FileReadTraceSource::READ_AHEAD: + return "read_ahead"; + case FileReadTraceSource::HOLE_FILL: + return "hole_fill"; + case FileReadTraceSource::NORMAL: + return context->reader_type == ReaderType::READER_QUERY ? "sync" : "other"; + } + __builtin_unreachable(); +} + +} // namespace + +ReadIOTraceWriter::ReadIOTraceWriter(std::string directory, size_t flush_bytes, + std::chrono::milliseconds flush_interval) + : _directory(std::move(directory)), + _flush_bytes(flush_bytes), + _flush_interval(flush_interval) { + DORIS_CHECK(_flush_bytes > 0); + DORIS_CHECK(_flush_interval.count() > 0); +} + +ReadIOTraceWriter::~ReadIOTraceWriter() { + stop(); +} + +void ReadIOTraceWriter::_drop(uint64_t events) { + _dropped_events += events; + read_trace_dropped_events << events; +} + +void ReadIOTraceWriter::append(std::string_view line) { + std::lock_guard lock(_mutex); + if (_stopping) { + _drop(1); + return; + } + if (_thread == nullptr) { + const auto status = Thread::create( + "io", "read_io_trace", [this] { _run(); }, &_thread); + if (!status.ok()) { + LOG(WARNING) << "failed to start read IO trace writer: " << status; + _stopping = true; + _drop(1); + return; + } + } + _pending.append(line); Review Comment: [P2] Bound the trace writer's buffered memory With tracing enabled, a slow or stalled output disk lets all read-ahead and hole-fill producers keep appending here while the writer blocks in `appendv()` outside the mutex. `_flush_bytes` only wakes that writer; it does not cap either `_pending` or the in-flight `batch`, and both are untracked `std::string` allocations outside the read-ahead budgets. After a burst drains, `batch.clear()` still retains its capacity, including after tracing is disabled. Thus diagnostic capture can exhaust BE memory despite bounded query buffers. Account for and cap retained trace bytes, report dropped events through the existing checkpoints, and release excessive burst capacity. ########## be/src/storage/segment/column_reader.cpp: ########## @@ -2852,6 +2856,83 @@ Status FileColumnIterator::init(const ColumnIteratorOptions& opts) { FileColumnIterator::~FileColumnIterator() = default; +Status FileColumnIterator::_init_read_ahead(const ColumnReadAheadRequest& request) { + request.sanity_check(); + if (_read_ahead != nullptr) { + DORIS_CHECK(_read_ahead->options() == request.options()); + DORIS_CHECK(_read_ahead->reverse() == request.reverse); + return Status::OK(); + } + + OrdinalIndexReader* ordinal_index = nullptr; + RETURN_IF_ERROR(_reader->get_ordinal_index_reader(ordinal_index, _opts.stats)); + DORIS_CHECK(ordinal_index != nullptr); + std::vector<ColumnReadAheadPage> pages; + pages.reserve(ordinal_index->num_data_pages()); Review Comment: [P2] Track the retained read-ahead page metadata Enabling read-ahead copies every physical page in each scalar column into this plain `std::vector`, retained as `ColumnReadAhead::_pages`; dense scans also retain `_candidate_pages` and window-map nodes. These allocations have neither a tracked allocator nor manual MemTracker accounting, and happen before scheduler admission, so they survive even when every range is rejected. On 64-bit targets the two page arrays alone require at least 44 bytes per selected page: wide segments and concurrent scanners can retain substantial memory that query limits never see. Use tracked containers or balanced accounting/reservation for the planning metadata, preferably sharing immutable ordinal metadata, and cover the budget-rejection case. ########## be/src/io/fs/file_range_read_scheduler.cpp: ########## @@ -0,0 +1,680 @@ +// 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/fs/file_range_read_scheduler.h" + +#include <exception> +#include <limits> +#include <utility> + +#include "common/exception.h" +#include "common/logging.h" +#include "core/allocator.h" +#include "cpp/sync_point.h" +#include "io/fs/read_ahead_metrics.h" +#include "io/fs/read_io_trace.h" +#include "runtime/memory/mem_tracker_limiter.h" +#include "runtime/thread_context.h" +#include "util/defer_op.h" +#include "util/threadpool.h" +#include "util/time.h" + +namespace doris::io { + +using FileRangeReadAllocator = Allocator<false, false, false, DefaultMemoryAllocator, true>; + +namespace { + +bool add_overflows(size_t left, size_t right) { + return right > std::numeric_limits<size_t>::max() - left; +} + +} // namespace + +FileRangeReadSubmitResult FileRangeReadSubmitResult::rejected(FileRangeReadRejectReason reason, + Status status) { + return FileRangeReadSubmitResult { + .reads = {}, + .reject_reason = reason, + .status = std::move(status), + }; +} + +FileRangeReadBudget::FileRangeReadBudget(size_t max_bytes) : _max_bytes(max_bytes) { + DORIS_CHECK(_max_bytes > 0); +} + +size_t FileRangeReadBudget::resident_bytes() const { + std::lock_guard lock(_mutex); + return _resident_bytes; +} + +bool FileRangeReadBudget::_can_reserve_unlocked(size_t bytes) const { + return bytes <= _max_bytes - _resident_bytes; +} + +void FileRangeReadBudget::_reserve_unlocked(size_t bytes) { + DORIS_CHECK(_can_reserve_unlocked(bytes)); + _resident_bytes += bytes; +} + +void FileRangeReadBudget::_release(size_t bytes) { + std::lock_guard lock(_mutex); + DORIS_CHECK(bytes <= _resident_bytes); + _resident_bytes -= bytes; +} + +FileRangeReadContext::FileRangeReadContext(size_t max_bytes) : _budget(max_bytes) {} + +void FileRangeReadContext::cancel() { + if (_cancelled.exchange(true, std::memory_order_acq_rel)) { + return; + } + + std::lock_guard lock(_reads_mutex); + for (const auto& weak_read : _reads) { + if (auto read = weak_read.lock(); read != nullptr) { + read->request_cancel(); + } + } + _reads.clear(); +} + +void FileRangeReadContext::_register_reads( + const std::vector<std::shared_ptr<FileRangeRead>>& reads) { + bool cancel_reads = false; + { + std::lock_guard lock(_reads_mutex); + cancel_reads = _cancelled.load(std::memory_order_acquire); + if (!cancel_reads) { + std::erase_if(_reads, [](const auto& read) { return read.expired(); }); + _reads.reserve(_reads.size() + reads.size()); + for (const auto& read : reads) { + // Partial admission leaves null slots for ranges without a buffer or task. + if (read != nullptr) { + _reads.emplace_back(read); + } + } + } + } + if (cancel_reads) { + for (const auto& read : reads) { + if (read != nullptr) { + read->request_cancel(); + } + } + } +} + +FileRangeReadReservation::FileRangeReadReservation( + std::shared_ptr<FileRangeReadContext> context, + std::shared_ptr<FileRangeReadBudget> global_budget, size_t bytes) noexcept + : _context(std::move(context)), _global_budget(std::move(global_budget)), _bytes(bytes) { + DORIS_CHECK(_context != nullptr); + DORIS_CHECK(_global_budget != nullptr); + DORIS_CHECK(_bytes > 0); +} + +FileRangeReadReservation::FileRangeReadReservation(FileRangeReadReservation&& other) noexcept + : _context(std::move(other._context)), + _global_budget(std::move(other._global_budget)), + _bytes(std::exchange(other._bytes, 0)) {} + +FileRangeReadReservation& FileRangeReadReservation::operator=( + FileRangeReadReservation&& other) noexcept { + if (this != &other) { + _reset(); + _context = std::move(other._context); + _global_budget = std::move(other._global_budget); + _bytes = std::exchange(other._bytes, 0); + } + return *this; +} + +FileRangeReadReservation::~FileRangeReadReservation() { + _reset(); +} + +void FileRangeReadReservation::_reset() { + if (_context == nullptr) { + return; + } + _context->_budget._release(_bytes); + _global_budget->_release(_bytes); + read_ahead_bvars().resident_bytes << -static_cast<int64_t>(_bytes); + _context.reset(); + _global_budget.reset(); + _bytes = 0; +} + +FileRangeRead::FileRangeRead(FileRange range, std::shared_ptr<MemTrackerLimiter> tracker, + FileRangeReadReservation reservation, + std::shared_ptr<ReadAheadStatistics> statistics) + : _range(range), + _trace_id(ReadIOTrace::next_id()), + _tracker(std::move(tracker)), + _reservation(std::move(reservation)), + _statistics(std::move(statistics)) { + FileRangeReadAllocator allocator; + _data = reinterpret_cast<char*>(allocator.alloc(_range.size)); +} + +FileRangeRead::~FileRangeRead() { + SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_tracker); + FileRangeReadAllocator allocator; + allocator.free(_data, _range.size); +} + +Status FileRangeRead::create(FileRange range, std::shared_ptr<MemTrackerLimiter> tracker, + FileRangeReadReservation reservation, + std::shared_ptr<FileRangeRead>* output, + std::shared_ptr<ReadAheadStatistics> statistics) { + DORIS_CHECK(range.size > 0); + DORIS_CHECK(tracker != nullptr); + DORIS_CHECK(output != nullptr); + + Status injected_status; + TEST_SYNC_POINT_CALLBACK("FileRangeRead::create:inject_failure", &injected_status); + if (!injected_status.ok()) { + return injected_status; + } + + SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(tracker); + try { + output->reset(new FileRangeRead(range, std::move(tracker), std::move(reservation), + std::move(statistics))); + } catch (const Exception& exception) { + return exception.to_status(); + } catch (const std::exception& exception) { + return Status::MemoryAllocFailed("allocate file range read buffer failed: {}", + exception.what()); + } + return Status::OK(); +} + +bool FileRangeRead::_is_terminal(State state) { + return state == State::READY || state == State::FAILED || state == State::CANCELLED; +} + +Status FileRangeRead::wait(bool* ready_on_entry) { + std::unique_lock lock(_mutex); + if (ready_on_entry != nullptr) { + *ready_on_entry = _state == State::READY; + } + _cv.wait(lock, [this]() { return _is_terminal(_state); }); + return _status; +} + +void FileRangeRead::request_cancel() { + bool publish_cancelled = false; + { + std::lock_guard lock(_mutex); + if (_is_terminal(_state)) { + return; + } + _cancel_requested = true; + publish_cancelled = _state == State::QUEUED; + if (publish_cancelled) { + _state = State::CANCELLED; + _status = Status::Cancelled("asynchronous file range read cancelled"); + } + } + if (publish_cancelled) { + _cv.notify_all(); + } +} + +FileRangeRead::State FileRangeRead::state() const { + std::lock_guard lock(_mutex); + return _state; +} + +Slice FileRangeRead::data() const { + return slice(0, _range.size); +} + +Slice FileRangeRead::slice(size_t buffer_offset, size_t size) const { + std::lock_guard lock(_mutex); + DORIS_CHECK(_state == State::READY); + DORIS_CHECK(buffer_offset <= _range.size); + DORIS_CHECK(size <= _range.size - buffer_offset); + return {_data + buffer_offset, size}; +} + +FileRangeReadStats FileRangeRead::stats() const { + std::lock_guard lock(_mutex); + return _stats; +} + +bool FileRangeRead::_mark_running() { + std::lock_guard lock(_mutex); + if (_state == State::CANCELLED) { + return false; + } + DORIS_CHECK(_state == State::QUEUED); + _state = State::RUNNING; + return true; +} + +bool FileRangeRead::_is_cancel_requested() const { + std::lock_guard lock(_mutex); + return _cancel_requested; +} + +void FileRangeRead::_publish_ready(FileRangeReadStats stats) { + _publish_from_running(State::READY, Status::OK(), std::move(stats)); +} + +void FileRangeRead::_publish_failed(Status status, FileRangeReadStats stats) { + DORIS_CHECK(!status.ok()); + _publish_from_running(State::FAILED, std::move(status), std::move(stats)); +} + +void FileRangeRead::_publish_submit_failure(Status status) { + DORIS_CHECK(!status.ok()); + { + std::lock_guard lock(_mutex); + if (_state == State::CANCELLED) { + return; + } + DORIS_CHECK(_state == State::QUEUED); + _state = State::FAILED; + _status = std::move(status); + } + _cv.notify_all(); +} + +void FileRangeRead::_publish_cancelled(FileRangeReadStats stats) { + _publish_from_running(State::CANCELLED, + Status::Cancelled("asynchronous file range read cancelled"), + std::move(stats)); +} + +void FileRangeRead::_publish_from_running(State state, Status status, FileRangeReadStats stats) { + { + std::lock_guard lock(_mutex); + DORIS_CHECK(_state == State::RUNNING); + DORIS_CHECK(_is_terminal(state)); + if (_cancel_requested) { + state = State::CANCELLED; + status = Status::Cancelled("asynchronous file range read cancelled"); + } + _state = state; + _status = std::move(status); + _stats = std::move(stats); + // Publish accounting before waking a consumer that may immediately close its scanner. + if (_statistics != nullptr) { + auto& outcome = state == State::READY ? _statistics->completed_ranges + : state == State::FAILED ? _statistics->failed_ranges + : _statistics->cancelled_ranges; + COUNTER_UPDATE(&outcome, 1); + COUNTER_UPDATE(&_statistics->queue_wait_time, _stats.queue_wait_ns); + COUNTER_UPDATE(&_statistics->read_time, _stats.read_ns); + COUNTER_UPDATE(&_statistics->source_bytes, static_cast<int64_t>(_stats.bytes_read)); + COUNTER_UPDATE(&_statistics->successful_source_bytes, + static_cast<int64_t>(_stats.successful_source_bytes)); + COUNTER_UPDATE(&_statistics->remote_bytes, _stats.file_cache.bytes_read_from_remote); + COUNTER_UPDATE(&_statistics->local_bytes, _stats.file_cache.bytes_read_from_local); + COUNTER_UPDATE(&_statistics->remote_requests, _stats.file_cache.num_remote_io_total); + } + auto& metrics = read_ahead_bvars(); + metrics.source_bytes << static_cast<int64_t>(_stats.bytes_read); + metrics.successful_source_bytes << static_cast<int64_t>(_stats.successful_source_bytes); + metrics.remote_bytes << _stats.file_cache.bytes_read_from_remote; + metrics.remote_requests << _stats.file_cache.num_remote_io_total; + if (state == State::FAILED) { + read_ahead_bvars().failed_ranges << 1; + } + } + _cv.notify_all(); +} + +FileRangeReadIOContext FileRangeReadIOContext::from_caller(const IOContext& source) { + FileRangeReadIOContext result; + result.io_context = source; + if (source.query_id != nullptr) { + result.query_id = std::make_shared<TUniqueId>(*source.query_id); + } + result.io_context.file_cache_stats = nullptr; + result.io_context.file_reader_stats = nullptr; + result.io_context.remote_scan_cache_write_limiter = nullptr; + result.io_context.condition_cache_filtered_rows = 0; + result.io_context.predicate_filtered_rows = 0; + result.io_context.cache_write_mode_override = CacheWriteMode::NO_WRITE; + result.io_context.query_id = result.query_id.get(); + return result; +} + +FileRangeReadScheduler::FileRangeReadScheduler(FileRangeReadSchedulerOptions options, + ThreadPool* executor) + : _options(options), + _global_budget(std::make_shared<FileRangeReadBudget>(options.max_bytes_per_be)), + _mem_tracker(MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::CACHE, + "FileRangeReadScheduler")), + _executor(executor) { + DORIS_CHECK(_executor != nullptr); + read_ahead_bvars().resident_bytes << 0; +} + +FileRangeReadScheduler::~FileRangeReadScheduler() { + shutdown(); +} + +Status FileRangeReadSchedulerOptions::validate() const { + if (max_bytes_per_query == 0) { + return Status::InvalidArgument("file range read query byte limit must be positive"); + } + if (max_bytes_per_be == 0) { + return Status::InvalidArgument("file range read BE byte limit must be positive"); + } + return Status::OK(); +} + +Status FileRangeReadScheduler::create(const FileRangeReadSchedulerOptions& options, + ThreadPool* executor, + std::unique_ptr<FileRangeReadScheduler>* output) { + DORIS_CHECK(output != nullptr); + DORIS_CHECK(executor != nullptr); + RETURN_IF_ERROR(options.validate()); + output->reset(new FileRangeReadScheduler(options, executor)); + return Status::OK(); +} + +std::shared_ptr<FileRangeReadContext> FileRangeReadScheduler::create_context() const { + return std::make_shared<FileRangeReadContext>(_options.max_bytes_per_query); +} + +FileRangeReadSubmitResult FileRangeReadScheduler::try_submit( + const std::vector<FileRange>& ranges, FileReaderSPtr reader, + FileRangeReadIOContext io_context, const std::shared_ptr<FileRangeReadContext>& context) { + const auto record_rejection = [&](FileRangeReadRejectReason reason) { + read_ahead_bvars().rejected_batches << 1; + if (io_context.statistics != nullptr) { + COUNTER_UPDATE(&io_context.statistics->rejected_batches, 1); + if (reason == FileRangeReadRejectReason::QUERY_BYTE_LIMIT) { + COUNTER_UPDATE(&io_context.statistics->query_budget_rejected_batches, 1); + } else if (reason == FileRangeReadRejectReason::GLOBAL_BYTE_LIMIT) { + COUNTER_UPDATE(&io_context.statistics->be_budget_rejected_batches, 1); + } + } + }; + const auto reject = [&](FileRangeReadRejectReason reason, Status status = Status::OK()) { + record_rejection(reason); + return FileRangeReadSubmitResult::rejected(reason, std::move(status)); + }; + Status validation_status = _validate_request(ranges, reader, context); + if (!validation_status.ok()) { + return reject(FileRangeReadRejectReason::INVALID_REQUEST, std::move(validation_status)); + } + if (!accepting()) { + return reject(FileRangeReadRejectReason::SHUTTING_DOWN, + Status::Cancelled("file range read scheduler is shutting down")); + } + if (context->cancelled()) { + return reject(FileRangeReadRejectReason::QUERY_CANCELLED, + Status::Cancelled("query cancelled before file range submission")); + } + + std::vector<std::optional<FileRangeReadReservation>> reservations(ranges.size()); + const auto reserve_reason = _reserve_batch(context, ranges, &reservations); + + std::vector<std::shared_ptr<FileRangeRead>> reads(ranges.size()); + std::vector<ReadTask> tasks; + tasks.reserve(ranges.size()); + for (size_t index = 0; index < ranges.size(); ++index) { + // Keep the input slot empty when admission skipped a range, so page mappings stay valid. + if (!reservations[index].has_value()) { + continue; + } + std::shared_ptr<FileRangeRead> read; + Status status = + FileRangeRead::create(ranges[index], _mem_tracker, std::move(*reservations[index]), + &read, io_context.statistics); + if (!status.ok()) { + return reject(FileRangeReadRejectReason::ALLOCATION_FAILED, std::move(status)); + } + tasks.push_back(ReadTask { + .reader = reader, .read = read, .context = context, .io_context = io_context}); + reads[index] = std::move(read); + } + // Every range may exceed the remaining budgets, leaving the whole request on the sync path. + if (tasks.empty()) { + DORIS_CHECK(reserve_reason != FileRangeReadRejectReason::NONE); + return reject(reserve_reason); + } + context->_register_reads(reads); + + FileRangeReadRejectReason submit_reason = FileRangeReadRejectReason::NONE; + Status submit_status = _submit_tasks(std::move(tasks), &submit_reason); + if (!submit_status.ok()) { + return reject(submit_reason, std::move(submit_status)); + } + // Record a partially budget-rejected batch once, after execution admission succeeds. If every + // executor task was rejected, the failure above already counted this submission. + if (reserve_reason != FileRangeReadRejectReason::NONE) { + record_rejection(reserve_reason); + } + return {.reads = std::move(reads), + .reject_reason = FileRangeReadRejectReason::NONE, + .status = Status::OK()}; +} + +Status FileRangeReadScheduler::_validate_request( + const std::vector<FileRange>& ranges, const FileReaderSPtr& reader, + const std::shared_ptr<FileRangeReadContext>& context) const { + if (reader == nullptr) { + return Status::InvalidArgument("file range reader is null"); + } + if (context == nullptr) { + return Status::InvalidArgument("file range read context is null"); + } + if (ranges.empty()) { + return Status::InvalidArgument("file range read batch is empty"); + } + + for (const auto& range : ranges) { + if (range.size == 0) { + return Status::InvalidArgument("file range at offset {} is empty", range.offset); + } + if (add_overflows(range.offset, range.size)) { + return Status::InvalidArgument("file range at offset {} with size {} overflows", + range.offset, range.size); + } + if (range.end() > reader->size()) { + return Status::InvalidArgument("file range [{}, {}) exceeds file size {}", range.offset, + range.end(), reader->size()); + } + } + return Status::OK(); +} + +FileRangeReadRejectReason FileRangeReadScheduler::_reserve_batch( + const std::shared_ptr<FileRangeReadContext>& context, const std::vector<FileRange>& ranges, + std::vector<std::optional<FileRangeReadReservation>>* reservations) { + DORIS_CHECK(reservations != nullptr); + DORIS_CHECK(reservations->size() == ranges.size()); + auto& query_budget = context->_budget; + std::scoped_lock lock(query_budget._mutex, _global_budget->_mutex); + + size_t query_remaining = query_budget._max_bytes - query_budget._resident_bytes; + size_t be_remaining = _global_budget->_max_bytes - _global_budget->_resident_bytes; + size_t reserved_bytes = 0; + auto reject_reason = FileRangeReadRejectReason::NONE; + for (size_t index = 0; index < ranges.size(); ++index) { + const size_t bytes = ranges[index].size; + const auto reason = bytes > query_remaining ? FileRangeReadRejectReason::QUERY_BYTE_LIMIT + : bytes > be_remaining ? FileRangeReadRejectReason::GLOBAL_BYTE_LIMIT + : FileRangeReadRejectReason::NONE; + // With 3 MiB available and ranges of 2, 2, and 1 MiB, admit the first and third ranges. + // A range that cannot fit must not prevent later, smaller ranges from using the budget. + if (reason != FileRangeReadRejectReason::NONE) { + if (reject_reason == FileRangeReadRejectReason::NONE) { + reject_reason = reason; + } + continue; + } + (*reservations)[index].emplace(FileRangeReadReservation(context, _global_budget, bytes)); + query_remaining -= bytes; + be_remaining -= bytes; + reserved_bytes += bytes; + } + + query_budget._reserve_unlocked(reserved_bytes); + _global_budget->_reserve_unlocked(reserved_bytes); + read_ahead_bvars().resident_bytes << static_cast<int64_t>(reserved_bytes); + return reject_reason; +} + +Status FileRangeReadScheduler::_submit_tasks(std::vector<ReadTask> tasks, + FileRangeReadRejectReason* reject_reason) { + DORIS_CHECK(!tasks.empty()); + DORIS_CHECK(reject_reason != nullptr); + std::lock_guard lock(_mutex); + if (!_accepting) { + *reject_reason = FileRangeReadRejectReason::SHUTTING_DOWN; + return Status::Cancelled("file range read scheduler is shutting down"); + } + + size_t submitted_tasks = 0; + Status first_failure; + for (auto& task : tasks) { + auto read = task.read; + DORIS_CHECK(_inflight_reads.emplace(read).second); + read_ahead_bvars().inflight_ranges << 1; + task.submit_ns = MonotonicNanos(); + Status status = _executor->submit_func([this, task = std::move(task), read]() mutable { + Defer task_finished {[this, read]() { _task_finished(read); }}; + _run_task(std::move(task)); + }); + if (status.ok()) { + ++submitted_tasks; + if (read->_statistics != nullptr) { + COUNTER_UPDATE(&read->_statistics->submitted_ranges, 1); + COUNTER_UPDATE(&read->_statistics->submitted_bytes, + static_cast<int64_t>(read->range().size)); + } + continue; + } + DORIS_CHECK(_inflight_reads.erase(read) == 1); Review Comment: [P2] Preserve callback ownership after worker creation fails With the supported `segment_prefetch_thread_pool_thread_num_min=0`, a transient failure creating the first worker can make `ThreadPool::do_submit()` return an error after it has already queued this callback; it does not roll that enqueue back. This branch removes the read from `_inflight_reads` and marks it `FAILED`. If the next range or batch successfully starts a worker, the retained callback runs and `_mark_running()` aborts because the state is no longer `QUEUED`. Dropping the scheduler can also leave that callback holding raw `this`. Ensure failed submission relinquishes the queued callback, or coordinate its execution/lifetime before finalizing the read. Cover one worker-creation failure followed by success, beyond the current capacity/shutdown rejection tests. -- 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]
