Gabriel39 commented on code in PR #66773:
URL: https://github.com/apache/doris/pull/66773#discussion_r3828928758
##########
be/src/io/fs/hdfs_file_reader.cpp:
##########
@@ -113,16 +111,14 @@ HdfsFileReader::~HdfsFileReader() {
}
Status HdfsFileReader::close() {
- bool expected = false;
- if (_closed.compare_exchange_strong(expected, true,
std::memory_order_acq_rel)) {
- DorisMetrics::instance()->hdfs_file_open_reading->increment(-1);
- }
+ _closed = true;
return Status::OK();
}
Status HdfsFileReader::read_at_impl(size_t offset, Slice result, size_t*
bytes_read,
const IOContext* io_ctx) {
SCOPED_TIMER(_total_read_time);
+ RETURN_IF_ERROR(_handle->ensure_open());
Review Comment:
Please validate `_handle` before invoking lazy open. Any error returned by
`do_read_at_impl()` clears `_handle` and destroys the accessor below, so a
subsequent `read_at()` reaches this line and dereferences null instead of
returning the existing `cached hdfs file handle has been destroyed` Status. The
same ordering also opens a never-opened file after `close()` before detecting
the closed state. Please move `ensure_open()` after the closed/null guards (or
add equivalent guards here) and add a regression test that performs a second
read after the first read fails.
##########
be/src/io/fs/file_handle_cache.cpp:
##########
@@ -26,44 +26,66 @@
#include <tuple>
#include "common/cast_set.h"
+#include "common/metrics/doris_metrics.h"
+#include "cpp/sync_point.h"
#include "io/fs/err_utils.h"
+#include "io/hdfs_util.h"
+#include "util/bvar_helper.h"
#include "util/hash_util.hpp"
#include "util/time.h"
namespace doris::io {
HdfsFileHandle::~HdfsFileHandle() {
if (_hdfs_file != nullptr && _fs != nullptr) {
VLOG_FILE << "hdfsCloseFile() fid=" << _hdfs_file;
- hdfsCloseFile(_fs, _hdfs_file); // TODO: check return code
+ SCOPED_BVAR_LATENCY(hdfs_bvar::hdfs_close_latency);
+ SYNC_POINT_HOOK_RETURN_VALUE(hdfsCloseFile(_fs, _hdfs_file),
+ "HdfsFileHandle::close::hdfsCloseFile");
+ DorisMetrics::instance()->hdfs_file_open_reading->increment(-1);
}
_fs = nullptr;
_hdfs_file = nullptr;
}
Status HdfsFileHandle::init(int64_t file_size) {
- _hdfs_file = hdfsOpenFile(_fs, _fname.c_str(), O_RDONLY, 0, 0, 0);
- if (_hdfs_file == nullptr) {
- std::string _err_msg = hdfs_error();
- // invoker maybe just skip Status.NotFound and continue
- // so we need distinguish between it and other kinds of errors
- if (_err_msg.find("No such file or directory") != std::string::npos) {
- return Status::NotFound(_err_msg);
- }
- return Status::InternalError("failed to open {}: {}", _fname,
_err_msg);
- }
-
_file_size = file_size;
if (_file_size <= 0) {
- hdfsFileInfo* file_info = hdfsGetPathInfo(_fs, _fname.c_str());
+ SCOPED_BVAR_LATENCY(hdfs_bvar::hdfs_get_path_info_latency);
+ auto* file_info = SYNC_POINT_HOOK_RETURN_VALUE(hdfsGetPathInfo(_fs,
_fname.c_str()),
+
"HdfsFileHandle::init::hdfsGetPathInfo");
if (file_info == nullptr) {
return Status::InternalError("failed to get file size of {}: {}",
_fname, hdfs_error());
}
_file_size = file_info->mSize;
+
TEST_SYNC_POINT_RETURN_WITH_VALUE("HdfsFileHandle::init::hdfsFreeFileInfo",
Status::OK());
hdfsFreeFileInfo(file_info, 1);
}
return Status::OK();
}
+Status HdfsFileHandle::ensure_open() {
+ std::call_once(_open_once, [this]() {
+ VLOG_DEBUG << "lazy open hdfs file: " << _fname;
+ SCOPED_BVAR_LATENCY(hdfs_bvar::hdfs_open_latency);
+ _hdfs_file =
+ SYNC_POINT_HOOK_RETURN_VALUE(hdfsOpenFile(_fs, _fname.c_str(),
O_RDONLY, 0, 0, 0),
+
"HdfsFileHandle::ensure_open::hdfsOpenFile");
+ if (_hdfs_file != nullptr) {
+ DorisMetrics::instance()->hdfs_file_open_reading->increment(1);
+ DorisMetrics::instance()->hdfs_file_reader_total->increment(1);
+ }
+ });
+ if (_hdfs_file == nullptr) {
+ std::string _err_msg = SYNC_POINT_HOOK_RETURN_VALUE(
Review Comment:
Please preserve the open result inside `call_once`. Only the thread running
the lambda calls `hdfsOpenFile`, while every caller executes `hdfs_error()`
here. libhdfs stores the last exception in the context of the current calling
thread, so waiting callers can see an empty or unrelated error and return
`InternalError` while the opening thread returned `NotFound`. Store the
`Status` or error text as handle state inside the lambda and return the same
result to every caller; please also add a concurrent open-failure test.
--
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]
