github-actions[bot] commented on code in PR #67292:
URL: https://github.com/apache/doris/pull/67292#discussion_r4344901080
##########
be/src/io/cache/async_cache_write_manager.cpp:
##########
@@ -493,6 +498,123 @@ bool
AsyncCacheWriteManager::try_submit(AsyncCacheWriteTask task) {
return true;
}
+AsyncCacheWriteBlockSubmitResult AsyncCacheWriteManager::try_submit_block(
+ AsyncCacheWriteBlockRequest request) {
+ DORIS_CHECK(request.data.data != nullptr);
+ DORIS_CHECK(request.data.size > 0);
+ DORIS_CHECK(request.buffer_size > 0);
+ DORIS_CHECK(request.data.size <= request.buffer_size);
+ if (!check_write_epoch(request.write_epoch)) {
+ return AsyncCacheWriteBlockSubmitResult::STALE_EPOCH;
+ }
+
+ AsyncCacheWriteBufferPtr buffer;
+ if (!allocate_tracked_buffer(request.buffer_size, &buffer).ok()) {
+ return AsyncCacheWriteBlockSubmitResult::BUFFER_ALLOCATION_FAILED;
+ }
+ std::memcpy(buffer->data(), request.data.data, request.data.size);
+
+ return try_submit_owned_block(AsyncCacheWriteOwnedBlockRequest {
+ .cache_hash = request.cache_hash,
+ .file_offset = request.file_offset,
+ .write_size = request.data.size,
+ .buffer = std::move(buffer),
+ .admission_ctx = std::move(request.admission_ctx),
+ .write_epoch = std::move(request.write_epoch),
+ .inflight_index = request.inflight_index,
+ });
+}
+
+AsyncCacheWriteBlockSubmitResult
AsyncCacheWriteManager::try_submit_owned_block(
+ AsyncCacheWriteOwnedBlockRequest request) {
+ DORIS_CHECK(request.buffer != nullptr);
+ DORIS_CHECK(request.write_size > 0);
+ DORIS_CHECK(request.write_size <= request.buffer->size());
+ if (!check_write_epoch(request.write_epoch)) {
+ return AsyncCacheWriteBlockSubmitResult::STALE_EPOCH;
+ }
+
+ const int64_t submit_ts_us = MonotonicMicros();
+ AsyncCacheWriteTask task {
+ .cache_hash = request.cache_hash,
+ .file_offset = request.file_offset,
+ .write_size = request.write_size,
+ .buffer = request.buffer,
+ .admission_ctx = std::move(request.admission_ctx),
+ .submit_ts_us = submit_ts_us,
+ .write_epoch = std::move(request.write_epoch),
+ .on_finalized = nullptr,
+ };
+ std::shared_ptr<InflightWriteBufferEntry> entry;
+ if (request.inflight_index != nullptr) {
+ entry = std::make_shared<InflightWriteBufferEntry>(request.buffer,
request.file_offset,
+ request.write_size,
submit_ts_us);
+ TEST_SYNC_POINT_CALLBACK(
+
"CachedRemoteFileReader::_submit_async_write_tasks:before_inflight_insert",
&task);
+ auto existing =
request.inflight_index->insert_if_absent(request.cache_hash,
+
request.file_offset, entry);
+ if (existing != nullptr) {
+ return AsyncCacheWriteBlockSubmitResult::ALREADY_INFLIGHT;
+ }
+ task.on_finalized = [cache_hash = request.cache_hash, offset =
request.file_offset,
+ inflight_index = request.inflight_index,
+ entry](const AsyncCacheWriteTask&) {
+ inflight_index->remove_if(cache_hash, offset, entry);
+ };
+ }
+
+ if (!try_submit(std::move(task), request.admission_mode)) {
+ if (entry != nullptr) {
+ request.inflight_index->remove_if(request.cache_hash,
request.file_offset, entry);
+ request.inflight_index->record_backpressure_rollback();
+ }
+ return AsyncCacheWriteBlockSubmitResult::REJECTED;
+ }
+ return AsyncCacheWriteBlockSubmitResult::SUBMITTED;
+}
+
+bool AsyncCacheWriteManager::can_accept_without_eviction(size_t buffer_size)
const {
+ return available_slots_without_eviction(buffer_size) > 0;
+}
+
+size_t AsyncCacheWriteManager::available_slots_without_eviction(size_t
buffer_size) const {
+ DORIS_CHECK(buffer_size > 0);
+ std::lock_guard lock(_queue_mutex);
+ if (!_started.load(std::memory_order_acquire) ||
!_accepting.load(std::memory_order_acquire)) {
+ return 0;
+ }
+ DORIS_CHECK(_task_buffer_size == 0 || _task_buffer_size == buffer_size);
+ const size_t max_pending_bytes =
_options.load(std::memory_order_acquire)->max_pending_bytes;
+ const size_t pending_bytes =
_pending_bytes.load(std::memory_order_relaxed);
+ if (pending_bytes >= max_pending_bytes) {
+ return 0;
+ }
+ return (max_pending_bytes - pending_bytes) / buffer_size;
+}
+
+bool AsyncCacheWriteManager::accepting() const {
+ return _started.load(std::memory_order_acquire) &&
_accepting.load(std::memory_order_acquire);
+}
+
+bool AsyncCacheWriteManager::should_skip_block_writeback(
+ const UInt128Wrapper& cache_hash, size_t offset, size_t size,
+ const CacheAdmissionContext& admission_ctx,
+ InflightWriteBufferIndex* inflight_index) const {
+ if (inflight_index != nullptr && inflight_index->lookup(cache_hash,
offset) != nullptr) {
+ return true;
+ }
+ ReadStatistics stats;
+ const auto context = admission_ctx.to_cache_context(&stats);
+ auto result = _cache->probe(cache_hash, offset, size, context);
Review Comment:
[P1] Handle irregular cached upload blocks before probing
Supported S3 upload settings can leave cache blocks that are not aligned to
the global block grid: `s3_write_buffer_size=5.5 MiB` with 1 MiB cache blocks
makes `S3FileWriter` populate intervals such as `[5,5.5)` and `[5.5,6.5)` MiB.
After the former is evicted, consuming a read-ahead range such as `[5.25,5.75)`
fetches remote bytes and submits a partial fragment for `[5,6)`. This call then
reaches `BlockFileCache::probe()`, whose `DORIS_CHECK` requires the resident
block to start at 5 MiB; finding the retained 5.5 MiB block aborts the BE.
Native queries with peer reads enabled previously used the synchronous,
interval-aware `get_or_set()` path even with async writes enabled, so enabling
read-ahead newly exposes this crash. Use an interval-aware coverage check here
and cover partially evicted, non-multiple S3 upload layouts.
--
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]
