Skip to content

Commit d07c9b3

Browse files
committed
[improvement](be) Default async cache writes to drop oldest
### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: The locked FIFO async file-cache writer supports both reject_new and drop_oldest, but the initial default retained reject_new. Make drop_oldest the production and service-option default so newly downloaded blocks remain available to adjacent reads under queue saturation, while keeping reject_new available through the dynamic configuration for rollback and A/B comparison. ### Release note `async_file_cache_write_queue_full_policy` now defaults to `drop_oldest`. Set it to `reject_new` to restore the previous admission behavior. ### Check List (For Author) - Test: No need to test (default-selection-only change; both policy implementations and runtime switching are covered by existing unit tests) - BE clang-format and check-format - Behavior changed: Yes. Full async write queues now replace the oldest queued task by default. - Does this need documentation: No
1 parent 6e82901 commit d07c9b3

4 files changed

Lines changed: 4 additions & 3 deletions

File tree

be/src/common/config.cpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1286,7 +1286,7 @@ DEFINE_mInt64(async_file_cache_write_max_pending_tasks_per_disk, "256");
12861286
DEFINE_mInt32(async_file_cache_write_batch_size, "16");
12871287
DEFINE_mInt64(async_file_cache_write_watchdog_warn_secs, "30");
12881288
DEFINE_mInt64(async_file_cache_write_watchdog_drop_secs, "120");
1289-
DEFINE_mString(async_file_cache_write_queue_full_policy, "reject_new");
1289+
DEFINE_mString(async_file_cache_write_queue_full_policy, "drop_oldest");
12901290
DEFINE_mBool(enable_async_file_cache_write_inflight_write_buffer_index, "true");
12911291
DEFINE_Int32(async_file_cache_write_inflight_write_buffer_index_shard_count, "64");
12921292
DEFINE_Validator(async_file_cache_write_workers_per_disk,

be/src/io/cache/async_cache_write_service.h

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -107,7 +107,7 @@ struct AsyncCacheWriteServiceOptions {
107107
size_t batch_size {1};
108108
int64_t watchdog_warn_secs {30};
109109
int64_t watchdog_drop_secs {120};
110-
AsyncCacheWriteQueueFullPolicy queue_full_policy {AsyncCacheWriteQueueFullPolicy::REJECT_NEW};
110+
AsyncCacheWriteQueueFullPolicy queue_full_policy {AsyncCacheWriteQueueFullPolicy::DROP_OLDEST};
111111
};
112112

113113
/// Owns the bounded async-write queue and workers for one BlockFileCache (one cache disk).

be/src/io/tools/async_file_cache_write_microbench.cpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -499,7 +499,7 @@ class BenchmarkEnvironment {
499499
config::async_file_cache_write_max_pending_tasks_per_disk =
500500
static_cast<int64_t>(std::max(FLAGS_reader_operations, FLAGS_service_operations));
501501
config::async_file_cache_write_batch_size = 16;
502-
config::async_file_cache_write_queue_full_policy = "reject_new";
502+
config::async_file_cache_write_queue_full_policy = "drop_oldest";
503503

504504
DORIS_CHECK(ExecEnv::GetInstance()->file_cache_factory() == nullptr);
505505
ExecEnv::GetInstance()->set_file_cache_open_fd_cache(std::make_unique<FDCache>());

be/test/io/cache/async_cache_write_service_test.cpp

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -122,6 +122,7 @@ TEST_F(AsyncCacheWriteServiceTest, TaskWritesDownloadedBlockAndCleansInflightEnt
122122
auto* index = cache->inflight_write_buffer_index();
123123
ASSERT_NE(service, nullptr);
124124
ASSERT_NE(index, nullptr);
125+
EXPECT_EQ(service->options().queue_full_policy, AsyncCacheWriteQueueFullPolicy::DROP_OLDEST);
125126
const uint64_t baseline_submitted = service->_submitted_metric->get_value();
126127
const uint64_t baseline_submitted_bytes = service->_submitted_bytes_metric->get_value();
127128
const uint64_t baseline_finished = service->_finished_metric->get_value();

0 commit comments

Comments
 (0)