diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp index c6ff22c626478e..5001ab84b183a2 100644 --- a/be/src/common/config.cpp +++ b/be/src/common/config.cpp @@ -1103,6 +1103,8 @@ DEFINE_Int32(ingest_binlog_work_pool_size, "-1"); // Download binlog rate limit, unit is KB/s, 0 means no limit DEFINE_Int32(download_binlog_rate_limit_kbs, "0"); +DEFINE_mInt32(buffered_reader_read_timeout_ms, "20000"); + // clang-format off #ifdef BE_TEST // test s3 diff --git a/be/src/common/config.h b/be/src/common/config.h index f621b442568ef3..2b7ff9367c1a2f 100644 --- a/be/src/common/config.h +++ b/be/src/common/config.h @@ -1159,6 +1159,8 @@ DECLARE_Int32(ingest_binlog_work_pool_size); // Download binlog rate limit, unit is KB/s DECLARE_Int32(download_binlog_rate_limit_kbs); +DECLARE_mInt32(buffered_reader_read_timeout_ms); + #ifdef BE_TEST // test s3 DECLARE_String(test_s3_resource); diff --git a/be/src/io/fs/buffered_reader.cpp b/be/src/io/fs/buffered_reader.cpp index fdcba04190f992..30a49f7fc95d7c 100644 --- a/be/src/io/fs/buffered_reader.cpp +++ b/be/src/io/fs/buffered_reader.cpp @@ -387,20 +387,15 @@ Status MergeRangeFileReader::_fill_box(int range_index, size_t start_offset, siz return Status::OK(); } -// the condition variable would wait at most 10 seconds -// otherwise it would quit the procedure and treat it -// as one time out error status and would make the load -// task failed -constexpr static int WAIT_TIME_OUT_MS = 10000; - // there exists occasions where the buffer is already closed but // some prior tasks are still queued in thread pool, so we have to check whether // the buffer is closed each time the condition variable is notified. void PrefetchBuffer::reset_offset(size_t offset) { { std::unique_lock lck {_lock}; - if (!_prefetched.wait_for(lck, std::chrono::milliseconds(WAIT_TIME_OUT_MS), - [this]() { return _buffer_status != BufferStatus::PENDING; })) { + if (!_prefetched.wait_for( + lck, std::chrono::milliseconds(config::buffered_reader_read_timeout_ms), + [this]() { return _buffer_status != BufferStatus::PENDING; })) { _prefetch_status = Status::TimedOut("time out when reset prefetch buffer"); return; } @@ -427,10 +422,12 @@ void PrefetchBuffer::reset_offset(size_t offset) { void PrefetchBuffer::prefetch_buffer() { { std::unique_lock lck {_lock}; - if (!_prefetched.wait_for(lck, std::chrono::milliseconds(WAIT_TIME_OUT_MS), [this]() { - return _buffer_status == BufferStatus::RESET || - _buffer_status == BufferStatus::CLOSED; - })) { + if (!_prefetched.wait_for( + lck, std::chrono::milliseconds(config::buffered_reader_read_timeout_ms), + [this]() { + return _buffer_status == BufferStatus::RESET || + _buffer_status == BufferStatus::CLOSED; + })) { _prefetch_status = Status::TimedOut("time out when invoking prefetch buffer"); return; } @@ -470,7 +467,8 @@ void PrefetchBuffer::prefetch_buffer() { _statis.prefetch_request_io += 1; _statis.prefetch_request_bytes += _len; std::unique_lock lck {_lock}; - if (!_prefetched.wait_for(lck, std::chrono::milliseconds(WAIT_TIME_OUT_MS), + if (!_prefetched.wait_for(lck, + std::chrono::milliseconds(config::buffered_reader_read_timeout_ms), [this]() { return _buffer_status == BufferStatus::PENDING; })) { _prefetch_status = Status::TimedOut("time out when invoking prefetch buffer"); return; @@ -551,10 +549,12 @@ Status PrefetchBuffer::read_buffer(size_t off, const char* out, size_t buf_len, { std::unique_lock lck {_lock}; // buffer must be prefetched or it's closed - if (!_prefetched.wait_for(lck, std::chrono::milliseconds(WAIT_TIME_OUT_MS), [this]() { - return _buffer_status == BufferStatus::PREFETCHED || - _buffer_status == BufferStatus::CLOSED; - })) { + if (!_prefetched.wait_for( + lck, std::chrono::milliseconds(config::buffered_reader_read_timeout_ms), + [this]() { + return _buffer_status == BufferStatus::PREFETCHED || + _buffer_status == BufferStatus::CLOSED; + })) { _prefetch_status = Status::TimedOut("time out when read prefetch buffer"); return _prefetch_status; } @@ -590,7 +590,8 @@ Status PrefetchBuffer::read_buffer(size_t off, const char* out, size_t buf_len, void PrefetchBuffer::close() { std::unique_lock lck {_lock}; // in case _reader still tries to write to the buf after we close the buffer - if (!_prefetched.wait_for(lck, std::chrono::milliseconds(WAIT_TIME_OUT_MS), + if (!_prefetched.wait_for(lck, + std::chrono::milliseconds(config::buffered_reader_read_timeout_ms), [this]() { return _buffer_status != BufferStatus::PENDING; })) { _prefetch_status = Status::TimedOut("time out when close prefetch buffer"); return;