You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by ga...@apache.org on 2026/04/29 12:10:42 UTC
(doris) branch tmp-branch-4.0.5 created (now 0726024926a)
This is an automated email from the ASF dual-hosted git repository.
gavinchou pushed a change to branch tmp-branch-4.0.5
in repository https://gitbox.apache.org/repos/asf/doris.git
at 0726024926a [fix](be) Poll packed file async close without blocking (#62938)
This branch includes the following new commits:
new 0726024926a [fix](be) Poll packed file async close without blocking (#62938)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 01/01: [fix](be) Poll packed file async close without blocking (#62938)
Posted by ga...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gavinchou pushed a commit to branch tmp-branch-4.0.5
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 0726024926af468c829c3bf9b79eee129c089e67
Author: Gavin Chou <ga...@selectdb.com>
AuthorDate: Wed Apr 29 20:08:47 2026 +0800
[fix](be) Poll packed file async close without blocking (#62938)
---
be/src/io/fs/file_writer.h | 10 +++++
be/src/io/fs/packed_file_manager.cpp | 14 ++----
be/src/io/fs/s3_file_writer.cpp | 55 ++++++++++++-----------
be/src/io/fs/s3_file_writer.h | 1 +
be/test/io/fs/packed_file_manager_test.cpp | 72 ++++++++++++++++++++++++++----
5 files changed, 105 insertions(+), 47 deletions(-)
diff --git a/be/src/io/fs/file_writer.h b/be/src/io/fs/file_writer.h
index 0cda2b519c4..de4fc6f577a 100644
--- a/be/src/io/fs/file_writer.h
+++ b/be/src/io/fs/file_writer.h
@@ -72,6 +72,16 @@ public:
// If there is no data appended, an empty file will be persisted.
virtual Status close(bool non_block = false) = 0;
+ // Non-blocking probe for a previous close(true).
+ // OK means close finished successfully. NeedSendAgain means close is still running.
+ // Other errors mean close finished with error or the writer does not support this API.
+ // NOTE: This method consumes the async close result when it is ready. The caller must
+ // use it as the only completion path for that async close; mixing it with close(false)
+ // or another try_finish_close consumer is not supported.
+ virtual Status try_finish_close() {
+ return Status::NotSupported("try_finish_close is not supported");
+ }
+
Status append(const Slice& data) { return appendv(&data, 1); }
virtual Status appendv(const Slice* data, size_t data_cnt) = 0;
diff --git a/be/src/io/fs/packed_file_manager.cpp b/be/src/io/fs/packed_file_manager.cpp
index 6bff9eff6f3..654f2d933fd 100644
--- a/be/src/io/fs/packed_file_manager.cpp
+++ b/be/src/io/fs/packed_file_manager.cpp
@@ -851,11 +851,6 @@ void PackedFileManager::process_uploading_packed_files() {
Status upload_status = finalize_packed_file_upload(packed_file->packed_file_path,
packed_file->writer.get());
- if (upload_status.is<ErrorCode::ALREADY_CLOSED>()) {
- record_ready_to_upload(packed_file);
- handle_success(packed_file);
- continue;
- }
if (!upload_status.ok()) {
handle_failure(packed_file, upload_status);
continue;
@@ -873,16 +868,13 @@ void PackedFileManager::process_uploading_packed_files() {
continue;
}
- if (packed_file->writer->state() != FileWriter::State::CLOSED) {
+ Status status = packed_file->writer->try_finish_close();
+ if (status.is<ErrorCode::NEED_SEND_AGAIN>()) {
continue;
}
- Status status = packed_file->writer->close(true);
- if (status.is<ErrorCode::ALREADY_CLOSED>()) {
- handle_success(packed_file);
- continue;
- }
if (status.ok()) {
+ handle_success(packed_file);
continue;
}
diff --git a/be/src/io/fs/s3_file_writer.cpp b/be/src/io/fs/s3_file_writer.cpp
index a1b72b8dc98..812bbdc5171 100644
--- a/be/src/io/fs/s3_file_writer.cpp
+++ b/be/src/io/fs/s3_file_writer.cpp
@@ -142,32 +142,9 @@ Status S3FileWriter::close(bool non_block) {
};
if (state() == State::CLOSED) {
- if (_async_close_pack != nullptr) {
- _st = _async_close_pack->future.get();
- _async_close_pack = nullptr;
- // Return the final close status so that a blocking close issued after
- // an async close observes the real result just like the legacy behavior.
- if (!non_block && _st.ok()) {
- record_close_latency();
- }
- return _st;
- }
- if (non_block) {
- if (_st.ok()) {
- record_close_latency();
- return Status::Error<ErrorCode::ALREADY_CLOSED>(
- "S3FileWriter already closed, file path {}, file key {}",
- _obj_storage_path_opts.path.native(), _obj_storage_path_opts.key);
- }
- return _st;
- }
- if (_st.ok()) {
- record_close_latency();
- return Status::Error<ErrorCode::ALREADY_CLOSED>(
- "S3FileWriter already closed, file path {}, file key {}",
- _obj_storage_path_opts.path.native(), _obj_storage_path_opts.key);
- }
- return _st;
+ return Status::InternalError("S3FileWriter already closed, file path {}, file key {}",
+ _obj_storage_path_opts.path.native(),
+ _obj_storage_path_opts.key);
}
if (state() == State::ASYNC_CLOSING) {
if (non_block) {
@@ -194,7 +171,6 @@ Status S3FileWriter::close(bool non_block) {
s3_file_writer_async_close_queuing << -1;
s3_file_writer_async_close_processing << 1;
_st = _close_impl();
- _state = State::CLOSED;
_async_close_pack->promise.set_value(_st);
s3_file_writer_async_close_processing << -1;
});
@@ -207,6 +183,31 @@ Status S3FileWriter::close(bool non_block) {
return _st;
}
+Status S3FileWriter::try_finish_close() {
+ if (state() == State::CLOSED) {
+ if (_async_close_pack != nullptr) {
+ if (_async_close_pack->future.wait_for(std::chrono::seconds(0)) !=
+ std::future_status::ready) {
+ return Status::NeedSendAgain("async close is not finished");
+ }
+ _st = _async_close_pack->future.get();
+ _async_close_pack = nullptr;
+ }
+ return _st;
+ }
+ if (state() != State::ASYNC_CLOSING) {
+ return Status::NotSupported("S3FileWriter is not async closing");
+ }
+ CHECK(_async_close_pack != nullptr);
+ if (_async_close_pack->future.wait_for(std::chrono::seconds(0)) != std::future_status::ready) {
+ return Status::NeedSendAgain("async close is not finished");
+ }
+ _st = _async_close_pack->future.get();
+ _async_close_pack = nullptr;
+ _state = State::CLOSED;
+ return _st;
+}
+
bool S3FileWriter::_complete_part_task_callback(Status s) {
bool ret = false;
if (!s.ok()) [[unlikely]] {
diff --git a/be/src/io/fs/s3_file_writer.h b/be/src/io/fs/s3_file_writer.h
index 38e68b14c4d..825dc75ce86 100644
--- a/be/src/io/fs/s3_file_writer.h
+++ b/be/src/io/fs/s3_file_writer.h
@@ -71,6 +71,7 @@ public:
}
Status close(bool non_block = false) override;
+ Status try_finish_close() override;
private:
Status _close_impl();
diff --git a/be/test/io/fs/packed_file_manager_test.cpp b/be/test/io/fs/packed_file_manager_test.cpp
index 645f0896f1a..37800bf1620 100644
--- a/be/test/io/fs/packed_file_manager_test.cpp
+++ b/be/test/io/fs/packed_file_manager_test.cpp
@@ -55,24 +55,20 @@ public:
void set_start_close_status(Status st) { _start_close_status = std::move(st); }
void complete_async_close() {
if (_state == State::ASYNC_CLOSING) {
- _state = State::CLOSED;
+ _async_close_ready = true;
}
}
+ size_t close_calls() const { return _close_calls; }
+ size_t try_finish_close_calls() const { return _try_finish_close_calls; }
size_t append_calls() const { return _append_calls; }
bool closed() const { return _state == State::CLOSED; }
size_t bytes_appended() const override { return _bytes_appended; }
const std::string& written_data() const { return _written; }
Status close(bool non_block = false) override {
+ ++_close_calls;
if (_state == State::CLOSED) {
- if (non_block) {
- if (_close_status.ok()) {
- return Status::Error<ErrorCode::ALREADY_CLOSED>(
- "MockFileWriter already closed: {}", _path.native());
- }
- return _close_status;
- }
return Status::Error<ErrorCode::ALREADY_CLOSED>("MockFileWriter already closed: {}",
_path.native());
}
@@ -82,11 +78,18 @@ public:
}
if (_state == State::ASYNC_CLOSING) {
- return Status::InternalError("Don't submit async close multi times");
+ if (non_block) {
+ return Status::InternalError("Don't submit async close multi times");
+ }
+ _async_close_ready = true;
+ _async_close_consumed = true;
+ _state = State::CLOSED;
+ return _close_status;
}
if (non_block) {
_state = State::ASYNC_CLOSING;
+ _async_close_ready = false;
return Status::OK();
}
@@ -94,6 +97,19 @@ public:
return _close_status;
}
+ Status try_finish_close() override {
+ ++_try_finish_close_calls;
+ if (_state == State::CLOSED) {
+ return _async_close_consumed ? _close_status : Status::OK();
+ }
+ if (_state != State::ASYNC_CLOSING || !_async_close_ready) {
+ return Status::NeedSendAgain("async close is not finished");
+ }
+ _state = State::CLOSED;
+ _async_close_consumed = true;
+ return _close_status;
+ }
+
Status appendv(const Slice* data, size_t data_cnt) override {
if (!_append_status.ok()) {
return _append_status;
@@ -114,10 +130,14 @@ private:
Path _path;
size_t _bytes_appended = 0;
size_t _append_calls = 0;
+ size_t _close_calls = 0;
+ size_t _try_finish_close_calls = 0;
std::string _written;
Status _start_close_status = Status::OK();
Status _append_status = Status::OK();
Status _close_status = Status::OK();
+ bool _async_close_ready = false;
+ bool _async_close_consumed = false;
State _state = State::OPENED;
};
@@ -581,6 +601,40 @@ TEST_F(PackedFileManagerTest, ProcessUploadingFilesSetsFailedWhenAsyncCloseFails
EXPECT_NE(failed->last_error.find("async close fail"), std::string::npos);
}
+TEST_F(PackedFileManagerTest, ProcessUploadingFilesPollsAsyncCloseWithoutBlocking) {
+ std::string payload = "abc";
+ Slice slice(payload);
+ auto info = default_append_info();
+ ASSERT_TRUE(manager->append_small_file("async_poll_fail", slice, info).ok());
+ ASSERT_TRUE(manager->mark_current_packed_file_for_upload(_resource_id).ok());
+ ASSERT_EQ(manager->uploading_packed_files_for_test().size(), 1);
+
+ auto uploading = manager->uploading_packed_files_for_test().begin()->second;
+ auto* writer = dynamic_cast<MockFileWriter*>(uploading->writer.get());
+ ASSERT_NE(writer, nullptr);
+ uploading->state = PackedFileManager::PackedFileState::UPLOADING;
+ writer->set_close_status(Status::IOError("async close poll fail"));
+ ASSERT_TRUE(writer->close(true).ok());
+ ASSERT_EQ(writer->close_calls(), 1);
+
+ manager->process_uploading_packed_files();
+ EXPECT_EQ(uploading->state.load(), PackedFileManager::PackedFileState::UPLOADING);
+ EXPECT_EQ(manager->uploading_packed_files_for_test().size(), 1);
+ EXPECT_EQ(manager->uploaded_packed_files_for_test().size(), 0);
+ EXPECT_EQ(writer->close_calls(), 1);
+ EXPECT_EQ(writer->try_finish_close_calls(), 1);
+
+ writer->complete_async_close();
+ manager->process_uploading_packed_files();
+ EXPECT_EQ(writer->close_calls(), 1);
+ EXPECT_EQ(writer->try_finish_close_calls(), 2);
+ EXPECT_EQ(manager->uploading_packed_files_for_test().size(), 0);
+ ASSERT_EQ(manager->uploaded_packed_files_for_test().size(), 1);
+ auto failed = manager->uploaded_packed_files_for_test().begin()->second;
+ EXPECT_EQ(failed->state.load(), PackedFileManager::PackedFileState::FAILED);
+ EXPECT_NE(failed->last_error.find("async close poll fail"), std::string::npos);
+}
+
TEST_F(PackedFileManagerTest, AppendPackedFileInfoToFileTail) {
std::string payload = "abc";
Slice slice(payload);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org