From af9ef98569142d9ab3fff358dd86ecc1fdd24c94 Mon Sep 17 00:00:00 2001 From: zchuango Date: Fri, 24 Jul 2026 14:28:30 +0800 Subject: [PATCH 1/4] add the progressive timeout reader --- example/http_c++/http_client.cpp | 34 ++++++++++ example/http_c++/http_server.cpp | 7 ++ src/brpc/controller.cpp | 93 ++++++++++++++++++++++++++- src/brpc/controller.h | 10 ++- src/brpc/errno.proto | 1 + src/brpc/policy/http_rpc_protocol.cpp | 1 + src/brpc/policy/http_rpc_protocol.h | 12 +++- src/brpc/progressive_reader.h | 2 + 8 files changed, 154 insertions(+), 6 deletions(-) diff --git a/example/http_c++/http_client.cpp b/example/http_c++/http_client.cpp index 23222dee9b..7df7461135 100644 --- a/example/http_c++/http_client.cpp +++ b/example/http_c++/http_client.cpp @@ -25,8 +25,11 @@ #include #include #include +#include "bthread/countdown_event.h" DEFINE_string(d, "", "POST this data to the http server"); +DEFINE_bool(progressive, false, "whether or not progressive read data from server"); +DEFINE_int32(progressive_read_timeout_ms, 5000, "progressive read data idle timeout in milliseconds"); DEFINE_string(load_balancer, "", "The algorithm for load balancing"); DEFINE_int32(timeout_ms, 2000, "RPC timeout in milliseconds"); DEFINE_int32(max_retry, 3, "Max retries(not including the first RPC)"); @@ -36,6 +39,25 @@ namespace brpc { DECLARE_bool(http_verbose); } +class PartDataReader: public brpc::ProgressiveReader { +public: + explicit PartDataReader(bthread::CountdownEvent* done): _done(done){} + + butil::Status OnReadOnePart(const void* data, size_t length) { + memcpy(_buffer, data, length); + LOG(INFO) << "data ： " << _buffer << " size : " << length; + return butil::Status::OK(); + } + + void OnEndOfMessage(const butil::Status& status) { + _done->signal(); + LOG(INFO) << "progressive read data final status : " << status; + } +private: + char _buffer[1024]; + bthread::CountdownEvent* _done; +}; + int main(int argc, char* argv[]) { // Parse gflags. We recommend you to use gflags as well. GFLAGS_NAMESPACE::ParseCommandLineFlags(&argc, &argv, true); @@ -71,6 +93,11 @@ int main(int argc, char* argv[]) { cntl.request_attachment().append(FLAGS_d); } + if (FLAGS_progressive) { + cntl.set_progressive_read_timeout_ms(FLAGS_progressive_read_timeout_ms); + cntl.response_will_be_read_progressively(); + } + // Because `done'(last parameter) is NULL, this function waits until // the response comes back or error occurs(including timedout). channel.CallMethod(NULL, &cntl, NULL, NULL, NULL); @@ -78,6 +105,13 @@ int main(int argc, char* argv[]) { std::cerr << cntl.ErrorText() << std::endl; return -1; } + + if (FLAGS_progressive) { + bthread::CountdownEvent done(1); + cntl.ReadProgressiveAttachmentBy(new PartDataReader(&done)); + done.wait(); + LOG(INFO) << "wait client progressive read done safely"; + } // If -http_verbose is on, brpc already prints the response to stderr. if (!brpc::FLAGS_http_verbose) { std::cout << cntl.response_attachment() << std::endl; diff --git a/example/http_c++/http_server.cpp b/example/http_c++/http_server.cpp index 05c9a0ee4c..3cc4c63f86 100644 --- a/example/http_c++/http_server.cpp +++ b/example/http_c++/http_server.cpp @@ -31,6 +31,7 @@ DEFINE_int32(idle_timeout_s, -1, "Connection will be closed if there is no " DEFINE_string(certificate, "cert.pem", "Certificate file path to enable SSL"); DEFINE_string(private_key, "key.pem", "Private key file path to enable SSL"); DEFINE_string(ciphers, "", "Cipher suite used for SSL connections"); +DEFINE_bool(enable_progressive_timeout, false, "whether or not trigger progressive write attachement data timeout"); namespace example { @@ -104,6 +105,9 @@ class FileServiceImpl : public FileService { // sleep a while to send another part. bthread_usleep(10000); + if (FLAGS_enable_progressive_timeout && i > 50) { + bthread_usleep(100000000UL); + } } return NULL; } @@ -194,6 +198,9 @@ class HttpSSEServiceImpl : public HttpSSEService { // sleep a while to send another part. bthread_usleep(10000 * 10); + if (FLAGS_enable_progressive_timeout && i > 50) { + bthread_usleep(100000000UL); + } } return NULL; } diff --git a/src/brpc/controller.cpp b/src/brpc/controller.cpp index 0bcfb4122d..4c2e33c52e 100644 --- a/src/brpc/controller.cpp +++ b/src/brpc/controller.cpp @@ -94,8 +94,9 @@ namespace brpc { DEFINE_bool(graceful_quit_on_sigterm, false, "Register SIGTERM handle func to quit graceful"); DEFINE_bool(graceful_quit_on_sighup, false, - "Register SIGHUP handle func to quit graceful"); - + "Register SIGHUP handle func to quit graceful"); +DEFINE_bool(log_idle_progressive_read_close, false, + "Print log when an idle progressive read is closed"); const IdlNames idl_single_req_single_res = { "req", "res" }; const IdlNames idl_single_req_multi_res = { "req", "" }; const IdlNames idl_multi_req_single_res = { "", "res" }; @@ -174,6 +175,80 @@ class IgnoreAllRead : public ProgressiveReader { void OnEndOfMessage(const butil::Status&) {} }; +class ProgressiveTimeoutReader : public ProgressiveReader { +public: + explicit ProgressiveTimeoutReader(SocketId id, int32_t read_timeout_ms, ProgressiveReader* reader): + _socket_id(id), + _read_timeout_ms(read_timeout_ms), + _reader(reader), + _timeout_id(0), + _is_read_timeout(false) { + AddIdleReadTimeoutMonitor(); + } + + ~ProgressiveTimeoutReader() { + if(_timeout_id > 0) { + bthread_timer_del(_timeout_id); + } + } + + butil::Status OnReadOnePart(const void* data, size_t length) { + return _reader->OnReadOnePart(data, length); + } + + void OnEndOfMessage(const butil::Status& status) { + if (_is_read_timeout) { + _reader->OnEndOfMessage(butil::Status(EPROGREADTIMEOUT, "The progressive read timeout")); + } else { + _reader->OnEndOfMessage(status); + } + if(_timeout_id > 0) { + bthread_timer_del(_timeout_id); + _timeout_id = 0; + } + } + +private: + static void HandleIdleProgressiveReader(void* arg) { + if(arg == nullptr){ + LOG(ERROR) << "Controller::HandleIdleProgressiveReader arg is null."; + return; + } + ProgressiveTimeoutReader* reader = static_cast(arg); + SocketUniquePtr s; + if (Socket::Address(reader->_socket_id, &s) != 0) { + LOG(ERROR) << "not found the socket id : " << reader->_socket_id; + return; + } + auto log_idle = FLAGS_log_idle_progressive_read_close; + reader->_is_read_timeout = true; + LOG_IF(INFO, log_idle) << "progressive read timeout socket id : " << reader->_socket_id + << " progressive read timeout us : " << reader->_read_timeout_ms; + if (s->parsing_context() != NULL) { + s->parsing_context()->Destroy(); + } + s->ReleaseReferenceIfIdle(0); + } + void AddIdleReadTimeoutMonitor() { + if (_read_timeout_ms <= 0) { + return; + } + bthread_timer_add(&_timeout_id, + butil::milliseconds_from_now(_read_timeout_ms), + HandleIdleProgressiveReader, + this + ); + } + +private: + SocketId _socket_id; + int32_t _read_timeout_ms; + ProgressiveReader* _reader; + // Timer registered to trigger progressive timeout event + bthread_timer_t _timeout_id; + butil::atomic _is_read_timeout; +}; + static IgnoreAllRead* s_ignore_all_read = NULL; static pthread_once_t s_ignore_all_read_once = PTHREAD_ONCE_INIT; static void CreateIgnoreAllRead() { s_ignore_all_read = new IgnoreAllRead; } @@ -261,6 +336,7 @@ void Controller::ResetPods() { _backup_request_ms = UNSET_MAGIC_NUM; _backup_request_policy = NULL; _connect_timeout_ms = UNSET_MAGIC_NUM; + _progressive_read_timeout_ms = UNSET_MAGIC_NUM; _real_timeout_ms = UNSET_MAGIC_NUM; _deadline_us = -1; _timeout_id = 0; @@ -336,6 +412,15 @@ void Controller::Call::Reset() { stream_user_data = NULL; } +void Controller::set_progressive_read_timeout_ms(int32_t progressive_read_timeout_ms){ + if(progressive_read_timeout_ms <= 0x7fffffff){ + _progressive_read_timeout_ms = progressive_read_timeout_ms; + } else { + _progressive_read_timeout_ms = 0x7fffffff; + LOG(WARNING) << "progressive_read_timeout_seconds is limited to 0x7fffffff"; + } +} + void Controller::set_timeout_ms(int64_t timeout_ms) { if (timeout_ms <= 0x7fffffff) { _timeout_ms = timeout_ms; @@ -1611,6 +1696,10 @@ void Controller::ReadProgressiveAttachmentBy(ProgressiveReader* r) { __FUNCTION__)); } add_flag(FLAGS_PROGRESSIVE_READER); + if (progressive_read_timeout_ms() > 0) { + auto reader = new ProgressiveTimeoutReader(_rpa->GetSocketId(), _progressive_read_timeout_ms, r); + return _rpa->ReadProgressiveAttachmentBy(reader); + } return _rpa->ReadProgressiveAttachmentBy(r); } diff --git a/src/brpc/controller.h b/src/brpc/controller.h index cb518706ed..bc46e597de 100644 --- a/src/brpc/controller.h +++ b/src/brpc/controller.h @@ -48,7 +48,6 @@ #include "brpc/grpc.h" #include "brpc/kvmap.h" #include "brpc/rpc_dump.h" - // EAUTH is defined in MAC #ifndef EAUTH #define EAUTH ERPCAUTH @@ -179,7 +178,6 @@ friend void policy::ProcessThriftRequest(InputMessageBase*); uint64_t log_id; std::string request_id; }; - public: Controller(); Controller(const Inheritable& parent_ctx); @@ -193,6 +191,9 @@ friend void policy::ProcessThriftRequest(InputMessageBase*); // Set/get timeout in milliseconds for the RPC call. Use // ChannelOptions.timeout_ms on unset. + void set_progressive_read_timeout_ms(int32_t progressive_read_timeout_ms); + int32_t progressive_read_timeout_ms() const { return _progressive_read_timeout_ms; } + void set_timeout_ms(int64_t timeout_ms); int64_t timeout_ms() const { return _timeout_ms; } @@ -339,7 +340,9 @@ friend void policy::ProcessThriftRequest(InputMessageBase*); // Make the RPC end when the HTTP response has complete headers and let // user read the remaining body by using ReadProgressiveAttachmentBy(). - void response_will_be_read_progressively() { add_flag(FLAGS_READ_PROGRESSIVELY); } + void response_will_be_read_progressively() { + add_flag(FLAGS_READ_PROGRESSIVELY); + } // Make the RPC end when the HTTP request has complete headers and let // user read the remaining body by using ReadProgressiveAttachmentBy(). void request_will_be_read_progressively() { add_flag(FLAGS_READ_PROGRESSIVELY); } @@ -878,6 +881,7 @@ friend void policy::ProcessThriftRequest(InputMessageBase*); int32_t _timeout_ms; int32_t _connect_timeout_ms; int32_t _backup_request_ms; + int32_t _progressive_read_timeout_ms; // Priority: `_backup_request_policy' > `_backup_request_ms'. BackupRequestPolicy* _backup_request_policy; // If this rpc call has retry/backup request,this var save the real timeout for current call diff --git a/src/brpc/errno.proto b/src/brpc/errno.proto index 26ffadc201..166d82dc4a 100644 --- a/src/brpc/errno.proto +++ b/src/brpc/errno.proto @@ -41,6 +41,7 @@ enum Errno { ESSL = 1016; // SSL related error EH2RUNOUTSTREAMS = 1017; // The H2 socket was run out of streams EREJECT = 1018; // The Request is rejected + EPROGREADTIMEOUT = 1019; // The Progressive read timeout // Errno caused by server EINTERNAL = 2001; // Internal Server Error diff --git a/src/brpc/policy/http_rpc_protocol.cpp b/src/brpc/policy/http_rpc_protocol.cpp index 8cbe06980f..3fb9408850 100644 --- a/src/brpc/policy/http_rpc_protocol.cpp +++ b/src/brpc/policy/http_rpc_protocol.cpp @@ -1201,6 +1201,7 @@ ParseResult ParseHttpMessage(butil::IOBuf *source, Socket *socket, LOG(FATAL) << "Fail to new HttpContext"; return MakeParseError(PARSE_ERROR_NO_RESOURCE); } + http_imsg->SetSocketId(socket->id()); // Parsing http is costly, parsing an incomplete http message from the // beginning repeatedly should be avoided, otherwise the cost may reach // O(n^2) in the worst case. Save incomplete http messages in sockets diff --git a/src/brpc/policy/http_rpc_protocol.h b/src/brpc/policy/http_rpc_protocol.h index bc8bd06593..2b2e9296ab 100644 --- a/src/brpc/policy/http_rpc_protocol.h +++ b/src/brpc/policy/http_rpc_protocol.h @@ -87,11 +87,20 @@ class HttpContext : public ReadableProgressiveAttachment , public InputMessageBase , public HttpMessage { public: + SocketId GetSocketId() override { + return _socket_id; + } + + void SetSocketId(SocketId id) { + _socket_id = id; + } + explicit HttpContext(bool read_body_progressively, HttpMethod request_method = HTTP_METHOD_GET) : InputMessageBase() , HttpMessage(read_body_progressively, request_method) - , _is_stage2(false) { + , _is_stage2(false) + , _socket_id(0) { // add one ref for Destroy butil::intrusive_ptr(this).detach(); } @@ -122,6 +131,7 @@ class HttpContext : public ReadableProgressiveAttachment private: bool _is_stage2; + SocketId _socket_id; }; // Implement functions required in protocol.h diff --git a/src/brpc/progressive_reader.h b/src/brpc/progressive_reader.h index 6f54ae68a7..860068e2e6 100644 --- a/src/brpc/progressive_reader.h +++ b/src/brpc/progressive_reader.h @@ -20,6 +20,7 @@ #define BRPC_PROGRESSIVE_READER_H #include "brpc/shared_object.h" +#include "brpc/socket.h" namespace brpc { @@ -84,6 +85,7 @@ class ReadableProgressiveAttachment : public SharedObject { // Any error occurred should destroy the reader by calling r->Destroy(). // r->Destroy() should be guaranteed to be called once and only once. virtual void ReadProgressiveAttachmentBy(ProgressiveReader* r) = 0; + virtual SocketId GetSocketId() = 0; }; } // namespace brpc From d4912598f68ed0fe86b34711449847a31795e209 Mon Sep 17 00:00:00 2001 From: zchuango Date: Fri, 31 Jul 2026 23:11:40 +0800 Subject: [PATCH 2/4] optimize the code format --- src/brpc/controller.cpp | 10 ++++++++-- src/brpc/controller.h | 2 ++ 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/src/brpc/controller.cpp b/src/brpc/controller.cpp index 4c2e33c52e..0b7f966539 100644 --- a/src/brpc/controller.cpp +++ b/src/brpc/controller.cpp @@ -192,11 +192,16 @@ class ProgressiveTimeoutReader : public ProgressiveReader { } } - butil::Status OnReadOnePart(const void* data, size_t length) { + butil::Status OnReadOnePart(const void* data, size_t length) override { + if (_timeout_id > 0) { + bthread_timer_del(_timeout_id); + _timeout_id = 0; + } + AddIdleReadTimeoutMonitor(); return _reader->OnReadOnePart(data, length); } - void OnEndOfMessage(const butil::Status& status) { + void OnEndOfMessage(const butil::Status& status) override { if (_is_read_timeout) { _reader->OnEndOfMessage(butil::Status(EPROGREADTIMEOUT, "The progressive read timeout")); } else { @@ -229,6 +234,7 @@ class ProgressiveTimeoutReader : public ProgressiveReader { } s->ReleaseReferenceIfIdle(0); } + void AddIdleReadTimeoutMonitor() { if (_read_timeout_ms <= 0) { return; diff --git a/src/brpc/controller.h b/src/brpc/controller.h index bc46e597de..74ebf6f020 100644 --- a/src/brpc/controller.h +++ b/src/brpc/controller.h @@ -48,6 +48,7 @@ #include "brpc/grpc.h" #include "brpc/kvmap.h" #include "brpc/rpc_dump.h" + // EAUTH is defined in MAC #ifndef EAUTH #define EAUTH ERPCAUTH @@ -178,6 +179,7 @@ friend void policy::ProcessThriftRequest(InputMessageBase*); uint64_t log_id; std::string request_id; }; + public: Controller(); Controller(const Inheritable& parent_ctx); From c014d00315850c99c9111eb33e6c4f4896a1e4d4 Mon Sep 17 00:00:00 2001 From: zchuango Date: Thu, 6 Aug 2026 20:20:44 +0800 Subject: [PATCH 3/4] WIP: progressive read timeout review --- example/http_c++/http_client.cpp | 9 +- example/http_c++/http_server.cpp | 2 +- src/brpc/controller.cpp | 284 ++++++++++++++++++----- src/brpc/policy/http_rpc_protocol.h | 2 +- src/brpc/progressive_reader.h | 2 +- test/brpc_http_rpc_protocol_unittest.cpp | 237 ++++++++++++++++++- 6 files changed, 466 insertions(+), 70 deletions(-) diff --git a/example/http_c++/http_client.cpp b/example/http_c++/http_client.cpp index 7df7461135..3a09186f84 100644 --- a/example/http_c++/http_client.cpp +++ b/example/http_c++/http_client.cpp @@ -22,6 +22,7 @@ // - Access www.foo.com // ./http_client www.foo.com +#include #include #include #include @@ -44,17 +45,17 @@ class PartDataReader: public brpc::ProgressiveReader { explicit PartDataReader(bthread::CountdownEvent* done): _done(done){} butil::Status OnReadOnePart(const void* data, size_t length) { - memcpy(_buffer, data, length); - LOG(INFO) << "data ： " << _buffer << " size : " << length; + const std::string part(static_cast(data), length); + LOG(INFO) << "data: " << part << " size: " << length; return butil::Status::OK(); } void OnEndOfMessage(const butil::Status& status) { - _done->signal(); LOG(INFO) << "progressive read data final status : " << status; + _done->signal(); + delete this; } private: - char _buffer[1024]; bthread::CountdownEvent* _done; }; diff --git a/example/http_c++/http_server.cpp b/example/http_c++/http_server.cpp index 3cc4c63f86..4c3c8722fd 100644 --- a/example/http_c++/http_server.cpp +++ b/example/http_c++/http_server.cpp @@ -31,7 +31,7 @@ DEFINE_int32(idle_timeout_s, -1, "Connection will be closed if there is no " DEFINE_string(certificate, "cert.pem", "Certificate file path to enable SSL"); DEFINE_string(private_key, "key.pem", "Private key file path to enable SSL"); DEFINE_string(ciphers, "", "Cipher suite used for SSL connections"); -DEFINE_bool(enable_progressive_timeout, false, "whether or not trigger progressive write attachement data timeout"); +DEFINE_bool(enable_progressive_timeout, false, "whether or not trigger progressive write attachment data timeout"); namespace example { diff --git a/src/brpc/controller.cpp b/src/brpc/controller.cpp index 0b7f966539..0003f8cf21 100644 --- a/src/brpc/controller.cpp +++ b/src/brpc/controller.cpp @@ -73,6 +73,7 @@ BAIDU_REGISTER_ERRNO(brpc::EEOF, "Got EOF"); BAIDU_REGISTER_ERRNO(brpc::EUNUSED, "The socket was not needed"); BAIDU_REGISTER_ERRNO(brpc::ESSL, "SSL related operation failed"); BAIDU_REGISTER_ERRNO(brpc::EH2RUNOUTSTREAMS, "The H2 socket was run out of streams"); +BAIDU_REGISTER_ERRNO(brpc::EPROGREADTIMEOUT, "Progressive read timed out"); BAIDU_REGISTER_ERRNO(brpc::EINTERNAL, "General internal error"); BAIDU_REGISTER_ERRNO(brpc::ERESPONSE, "Bad response"); @@ -175,84 +176,224 @@ class IgnoreAllRead : public ProgressiveReader { void OnEndOfMessage(const butil::Status&) {} }; +struct ProgressiveReadTimeoutTask; + +struct ProgressiveReadTimeoutState { + ProgressiveReadTimeoutState(SocketId id, int32_t timeout_ms) + : socket_id(id) + , read_timeout_ms(timeout_ms) + , deadline_us(butil::cpuwide_time_us() + timeout_ms * 1000L) + , timer_id(0) + , timer_task(NULL) + , user_callback_running(false) + , reader_failed(false) + , timeout_triggered(false) + , end_delivered(false) {} + + butil::Mutex mutex; + const SocketId socket_id; + const int32_t read_timeout_ms; + int64_t deadline_us; + bthread_timer_t timer_id; + ProgressiveReadTimeoutTask* timer_task; + bool user_callback_running; + bool reader_failed; + bool timeout_triggered; + bool end_delivered; + butil::Status timer_error; +}; + +struct ProgressiveReadTimeoutTask { + explicit ProgressiveReadTimeoutTask( + const std::shared_ptr& state_in) + : state(state_in) {} + + std::shared_ptr state; +}; + class ProgressiveTimeoutReader : public ProgressiveReader { public: - explicit ProgressiveTimeoutReader(SocketId id, int32_t read_timeout_ms, ProgressiveReader* reader): - _socket_id(id), - _read_timeout_ms(read_timeout_ms), - _reader(reader), - _timeout_id(0), - _is_read_timeout(false) { - AddIdleReadTimeoutMonitor(); - } + ProgressiveTimeoutReader(SocketId id, int32_t read_timeout_ms, + ProgressiveReader* reader) + : _reader(reader) + , _state(new ProgressiveReadTimeoutState(id, read_timeout_ms)) {} - ~ProgressiveTimeoutReader() { - if(_timeout_id > 0) { - bthread_timer_del(_timeout_id); - } + int Start() { + std::unique_lock mu(_state->mutex); + return AddWatchdogLocked(_state, _state->read_timeout_ms * 1000L); } butil::Status OnReadOnePart(const void* data, size_t length) override { - if (_timeout_id > 0) { - bthread_timer_del(_timeout_id); - _timeout_id = 0; + { + std::unique_lock mu(_state->mutex); + if (_state->timeout_triggered) { + return MakeTimeoutStatus(_state->read_timeout_ms); + } + if (!_state->timer_error.ok()) { + return _state->timer_error; + } + _state->user_callback_running = true; + } + + butil::Status status = _reader->OnReadOnePart(data, length); + { + std::unique_lock mu(_state->mutex); + _state->user_callback_running = false; + if (_state->timeout_triggered) { + status = MakeTimeoutStatus(_state->read_timeout_ms); + } else if (!_state->timer_error.ok()) { + status = _state->timer_error; + } else if (status.ok() && !_state->end_delivered) { + _state->deadline_us = butil::cpuwide_time_us() + + _state->read_timeout_ms * 1000L; + } else if (!status.ok()) { + _state->reader_failed = true; + } } - AddIdleReadTimeoutMonitor(); - return _reader->OnReadOnePart(data, length); + return status; } void OnEndOfMessage(const butil::Status& status) override { - if (_is_read_timeout) { - _reader->OnEndOfMessage(butil::Status(EPROGREADTIMEOUT, "The progressive read timeout")); - } else { - _reader->OnEndOfMessage(status); - } - if(_timeout_id > 0) { - bthread_timer_del(_timeout_id); - _timeout_id = 0; + bthread_timer_t timer_id = 0; + ProgressiveReadTimeoutTask* timer_task = NULL; + butil::Status final_status = status; + ProgressiveReader* reader = NULL; + { + std::unique_lock mu(_state->mutex); + if (_state->end_delivered) { + LOG(ERROR) << "ProgressiveReader::OnEndOfMessage was called more than once"; + return; + } + _state->end_delivered = true; + timer_id = _state->timer_id; + timer_task = _state->timer_task; + _state->timer_id = 0; + _state->timer_task = NULL; + if (_state->timeout_triggered) { + final_status = MakeTimeoutStatus(_state->read_timeout_ms); + } else if (!_state->timer_error.ok()) { + final_status = _state->timer_error; + } + reader = _reader; + _reader = NULL; } + + CancelWatchdog(timer_id, timer_task); + reader->OnEndOfMessage(final_status); + delete this; } private: - static void HandleIdleProgressiveReader(void* arg) { - if(arg == nullptr){ - LOG(ERROR) << "Controller::HandleIdleProgressiveReader arg is null."; + ~ProgressiveTimeoutReader() override {} + + static butil::Status MakeTimeoutStatus(int32_t timeout_ms) { + return butil::Status( + EPROGREADTIMEOUT, + "Progressive read timed out after %d ms", timeout_ms); + } + + static butil::Status MakeTimerErrorStatus(int error_code) { + return butil::Status( + error_code, "Fail to add progressive read timeout timer: %s", + berror(error_code)); + } + + static void CancelWatchdog( + bthread_timer_t timer_id, ProgressiveReadTimeoutTask* timer_task) { + if (timer_id == 0) { return; } - ProgressiveTimeoutReader* reader = static_cast(arg); - SocketUniquePtr s; - if (Socket::Address(reader->_socket_id, &s) != 0) { - LOG(ERROR) << "not found the socket id : " << reader->_socket_id; - return; + const int rc = bthread_timer_del(timer_id); + if (rc == 0) { + delete timer_task; + } else if (rc == 1 || rc == EINVAL) { + // The callback owns timer_task once it starts running. EINVAL means + // that the callback has already finished and released the task. + } else { + LOG(ERROR) << "Unexpected bthread_timer_del error=" << rc; + } + } + + static int AddWatchdogLocked( + const std::shared_ptr& state, + int64_t delay_us) { + if (state->end_delivered || state->reader_failed) { + return ECANCELED; } - auto log_idle = FLAGS_log_idle_progressive_read_close; - reader->_is_read_timeout = true; - LOG_IF(INFO, log_idle) << "progressive read timeout socket id : " << reader->_socket_id - << " progressive read timeout us : " << reader->_read_timeout_ms; - if (s->parsing_context() != NULL) { - s->parsing_context()->Destroy(); + if (delay_us <= 0) { + delay_us = 1; } - s->ReleaseReferenceIfIdle(0); + ProgressiveReadTimeoutTask* task = + new (std::nothrow) ProgressiveReadTimeoutTask(state); + if (task == NULL) { + return ENOMEM; + } + bthread_timer_t timer_id = 0; + const int rc = bthread_timer_add( + &timer_id, butil::microseconds_from_now(delay_us), + HandleIdleProgressiveReader, task); + if (rc != 0) { + delete task; + return rc; + } + state->timer_id = timer_id; + state->timer_task = task; + return 0; } - void AddIdleReadTimeoutMonitor() { - if (_read_timeout_ms <= 0) { + static void HandleIdleProgressiveReader(void* arg) { + std::unique_ptr task( + static_cast(arg)); + const std::shared_ptr state = task->state; + bool fail_socket = false; + int error_code = 0; + std::string error_text; + { + std::unique_lock mu(state->mutex); + if (state->timer_task == task.get()) { + state->timer_id = 0; + state->timer_task = NULL; + } + if (state->end_delivered || state->reader_failed) { + return; + } + + const int64_t now_us = butil::cpuwide_time_us(); + if (state->user_callback_running || now_us < state->deadline_us) { + const int64_t delay_us = state->user_callback_running + ? state->read_timeout_ms * 1000L + : state->deadline_us - now_us; + const int rc = AddWatchdogLocked(state, delay_us); + if (rc != 0) { + state->timer_error = MakeTimerErrorStatus(rc); + fail_socket = true; + error_code = rc; + error_text = state->timer_error.error_str(); + } + } else { + state->timeout_triggered = true; + fail_socket = true; + error_code = EPROGREADTIMEOUT; + error_text = MakeTimeoutStatus(state->read_timeout_ms).error_str(); + } + } + + if (!fail_socket) { return; } - bthread_timer_add(&_timeout_id, - butil::milliseconds_from_now(_read_timeout_ms), - HandleIdleProgressiveReader, - this - ); + SocketUniquePtr socket; + if (Socket::Address(state->socket_id, &socket) != 0) { + LOG(ERROR) << "Fail to address socket_id=" << state->socket_id + << " after progressive read timeout"; + } else { + LOG_IF(INFO, FLAGS_log_idle_progressive_read_close) + << error_text << ", socket_id=" << state->socket_id; + socket->SetFailed(error_code, "%s", error_text.c_str()); + } } -private: - SocketId _socket_id; - int32_t _read_timeout_ms; ProgressiveReader* _reader; - // Timer registered to trigger progressive timeout event - bthread_timer_t _timeout_id; - butil::atomic _is_read_timeout; + const std::shared_ptr _state; }; static IgnoreAllRead* s_ignore_all_read = NULL; @@ -418,13 +559,9 @@ void Controller::Call::Reset() { stream_user_data = NULL; } -void Controller::set_progressive_read_timeout_ms(int32_t progressive_read_timeout_ms){ - if(progressive_read_timeout_ms <= 0x7fffffff){ - _progressive_read_timeout_ms = progressive_read_timeout_ms; - } else { - _progressive_read_timeout_ms = 0x7fffffff; - LOG(WARNING) << "progressive_read_timeout_seconds is limited to 0x7fffffff"; - } +void Controller::set_progressive_read_timeout_ms( + int32_t progressive_read_timeout_ms) { + _progressive_read_timeout_ms = progressive_read_timeout_ms; } void Controller::set_timeout_ms(int64_t timeout_ms) { @@ -1703,8 +1840,31 @@ void Controller::ReadProgressiveAttachmentBy(ProgressiveReader* r) { } add_flag(FLAGS_PROGRESSIVE_READER); if (progressive_read_timeout_ms() > 0) { - auto reader = new ProgressiveTimeoutReader(_rpa->GetSocketId(), _progressive_read_timeout_ms, r); - return _rpa->ReadProgressiveAttachmentBy(reader); + const SocketId socket_id = _rpa->GetSocketId(); + if (socket_id == INVALID_SOCKET_ID) { + pthread_once(&s_ignore_all_read_once, CreateIgnoreAllRead); + _rpa->ReadProgressiveAttachmentBy(s_ignore_all_read); + return r->OnEndOfMessage(butil::Status( + ENOTSUP, + "Progressive read timeout is only supported for HTTP/1.x")); + } + ProgressiveTimeoutReader* reader = new (std::nothrow) + ProgressiveTimeoutReader( + socket_id, _progressive_read_timeout_ms, r); + if (reader == NULL) { + pthread_once(&s_ignore_all_read_once, CreateIgnoreAllRead); + _rpa->ReadProgressiveAttachmentBy(s_ignore_all_read); + return r->OnEndOfMessage( + butil::Status(ENOMEM, "Fail to create progressive timeout reader")); + } + const int rc = reader->Start(); + if (rc != 0) { + pthread_once(&s_ignore_all_read_once, CreateIgnoreAllRead); + _rpa->ReadProgressiveAttachmentBy(s_ignore_all_read); + return reader->OnEndOfMessage(butil::Status( + rc, "Fail to add progressive read timeout timer: %s", berror(rc))); + } + return _rpa->ReadProgressiveAttachmentBy(reader); } return _rpa->ReadProgressiveAttachmentBy(r); } diff --git a/src/brpc/policy/http_rpc_protocol.h b/src/brpc/policy/http_rpc_protocol.h index 2b2e9296ab..cd41798e9f 100644 --- a/src/brpc/policy/http_rpc_protocol.h +++ b/src/brpc/policy/http_rpc_protocol.h @@ -100,7 +100,7 @@ class HttpContext : public ReadableProgressiveAttachment : InputMessageBase() , HttpMessage(read_body_progressively, request_method) , _is_stage2(false) - , _socket_id(0) { + , _socket_id(INVALID_SOCKET_ID) { // add one ref for Destroy butil::intrusive_ptr(this).detach(); } diff --git a/src/brpc/progressive_reader.h b/src/brpc/progressive_reader.h index 860068e2e6..c84be8b7e7 100644 --- a/src/brpc/progressive_reader.h +++ b/src/brpc/progressive_reader.h @@ -20,7 +20,7 @@ #define BRPC_PROGRESSIVE_READER_H #include "brpc/shared_object.h" -#include "brpc/socket.h" +#include "brpc/socket_id.h" namespace brpc { diff --git a/test/brpc_http_rpc_protocol_unittest.cpp b/test/brpc_http_rpc_protocol_unittest.cpp index 87837abf1a..73e9b3dcab 100644 --- a/test/brpc_http_rpc_protocol_unittest.cpp +++ b/test/brpc_http_rpc_protocol_unittest.cpp @@ -19,6 +19,7 @@ // Date: Sun Jul 13 15:04:18 CST 2014 +#include #include #include #include @@ -736,9 +737,13 @@ static void CopyPAPrefixedWithSeqNo(char* buf, uint64_t seq_no) { class DownloadServiceImpl : public ::test::DownloadService { public: DownloadServiceImpl(DonePlace done_place = DONE_BEFORE_CREATE_PA, - size_t num_repeat = 1) + size_t num_repeat = 1, + int write_interval_us = 0, + int initial_write_delay_us = 0) : _done_place(done_place) , _nrep(num_repeat) + , _write_interval_us(write_interval_us) + , _initial_write_delay_us(initial_write_delay_us) , _nwritten(0) , _ever_full(false) , _last_errno(0) {} @@ -762,6 +767,9 @@ class DownloadServiceImpl : public ::test::DownloadService { if (_done_place == DONE_BEFORE_CREATE_PA) { done_guard.reset(NULL); } + if (_initial_write_delay_us > 0) { + bthread_usleep(_initial_write_delay_us); + } ASSERT_GT(PA_DATA_LEN, 8u); // long enough to hold a 64-bit decimal. char buf[PA_DATA_LEN]; for (size_t c = 0; c < _nrep;) { @@ -778,6 +786,9 @@ class DownloadServiceImpl : public ::test::DownloadService { } } else { _nwritten += PA_DATA_LEN; + if (_write_interval_us > 0) { + bthread_usleep(_write_interval_us); + } } ++c; } @@ -840,6 +851,8 @@ class DownloadServiceImpl : public ::test::DownloadService { private: DonePlace _done_place; size_t _nrep; + int _write_interval_us; + int _initial_write_delay_us; size_t _nwritten; bool _ever_full; int _last_errno; @@ -941,6 +954,47 @@ class ReadBody : public brpc::ProgressiveReader, butil::Status _destroying_st; }; +class TimeoutReadBody : public brpc::ProgressiveReader, + public brpc::SharedObject { +public: + explicit TimeoutReadBody(int read_delay_us = 0, int read_error = 0) + : _read_delay_us(read_delay_us) + , _read_error(read_error) + , _nread(0) + , _nend(0) + , _end_error(0) { + butil::intrusive_ptr(this).detach(); + } + + butil::Status OnReadOnePart(const void*, size_t length) override { + if (_read_delay_us > 0) { + bthread_usleep(_read_delay_us); + } + _nread.fetch_add(length); + if (_read_error != 0) { + return butil::Status(_read_error, "intended progressive read failure"); + } + return butil::Status::OK(); + } + + void OnEndOfMessage(const butil::Status& status) override { + _end_error.store(status.error_code()); + _nend.fetch_add(1); + butil::intrusive_ptr(this, false); + } + + size_t read_bytes() const { return _nread.load(); } + int end_count() const { return _nend.load(); } + int end_error() const { return _end_error.load(); } + +private: + const int _read_delay_us; + const int _read_error; + std::atomic _nread; + std::atomic _nend; + std::atomic _end_error; +}; + #ifdef BUTIL_USE_ASAN static const int GENERAL_DELAY_US = 1000000; // 1s #else @@ -1034,6 +1088,187 @@ TEST_F(HttpTest, read_short_body_progressively) { } } +TEST_F(HttpTest, progressive_read_timeout_keeps_active_reader_alive) { + const int port = 8923; + brpc::Server server; + DownloadServiceImpl svc(DONE_BEFORE_CREATE_PA, 10000); + ASSERT_EQ(0, server.AddService(&svc, brpc::SERVER_DOESNT_OWN_SERVICE)); + ASSERT_EQ(0, server.Start(port, NULL)); + + brpc::Channel channel; + brpc::ChannelOptions options; + options.protocol = brpc::PROTOCOL_HTTP; + ASSERT_EQ(0, channel.Init(butil::EndPoint(butil::my_ip(), port), &options)); + + brpc::Controller cntl; + cntl.response_will_be_read_progressively(); + cntl.set_progressive_read_timeout_ms(1000); + cntl.http_request().uri() = "/DownloadService/Download"; + channel.CallMethod(NULL, &cntl, NULL, NULL, NULL); + ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText(); + + butil::intrusive_ptr reader(new TimeoutReadBody); + cntl.ReadProgressiveAttachmentBy(reader.get()); + for (int i = 0; i < 100 && reader->end_count() == 0; ++i) { + bthread_usleep(10000); + } + ASSERT_EQ(1, reader->end_count()); + EXPECT_EQ(0, reader->end_error()); + EXPECT_EQ(10000 * PA_DATA_LEN, reader->read_bytes()); +} + +TEST_F(HttpTest, progressive_read_timeout_closes_idle_http1_reader_once) { + const int port = 8923; + brpc::Server server; + DownloadServiceImpl svc(DONE_BEFORE_CREATE_PA, 2, 300000); + ASSERT_EQ(0, server.AddService(&svc, brpc::SERVER_DOESNT_OWN_SERVICE)); + ASSERT_EQ(0, server.Start(port, NULL)); + + butil::intrusive_ptr reader(new TimeoutReadBody); + { + brpc::Channel channel; + brpc::ChannelOptions options; + options.protocol = brpc::PROTOCOL_HTTP; + ASSERT_EQ(0, channel.Init(butil::EndPoint(butil::my_ip(), port), &options)); + { + brpc::Controller cntl; + cntl.response_will_be_read_progressively(); + cntl.set_progressive_read_timeout_ms(50); + cntl.http_request().uri() = "/DownloadService/Download"; + channel.CallMethod(NULL, &cntl, NULL, NULL, NULL); + ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText(); + cntl.ReadProgressiveAttachmentBy(reader.get()); + bthread_usleep(400000); + ASSERT_NE(0, svc.last_errno()); + EXPECT_EQ(0, reader->end_count()); + } + } + for (int i = 0; i < 100 && reader->end_count() == 0; ++i) { + bthread_usleep(10000); + } + ASSERT_EQ(1, reader->end_count()); + EXPECT_EQ(brpc::EPROGREADTIMEOUT, reader->end_error()); + bthread_usleep(400000); + EXPECT_EQ(1, reader->end_count()); +} + +TEST_F(HttpTest, progressive_read_timeout_before_first_body_part) { + const int port = 8923; + brpc::Server server; + DownloadServiceImpl svc(DONE_BEFORE_CREATE_PA, 1, 0, 300000); + ASSERT_EQ(0, server.AddService(&svc, brpc::SERVER_DOESNT_OWN_SERVICE)); + ASSERT_EQ(0, server.Start(port, NULL)); + + butil::intrusive_ptr reader(new TimeoutReadBody); + { + brpc::Channel channel; + brpc::ChannelOptions options; + options.protocol = brpc::PROTOCOL_HTTP; + ASSERT_EQ(0, channel.Init(butil::EndPoint(butil::my_ip(), port), &options)); + { + brpc::Controller cntl; + cntl.response_will_be_read_progressively(); + cntl.set_progressive_read_timeout_ms(50); + cntl.http_request().uri() = "/DownloadService/Download"; + channel.CallMethod(NULL, &cntl, NULL, NULL, NULL); + ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText(); + cntl.ReadProgressiveAttachmentBy(reader.get()); + bthread_usleep(400000); + ASSERT_NE(0, svc.last_errno()); + EXPECT_EQ(size_t(0), reader->read_bytes()); + EXPECT_EQ(0, reader->end_count()); + } + } + for (int i = 0; i < 100 && reader->end_count() == 0; ++i) { + bthread_usleep(10000); + } + ASSERT_EQ(1, reader->end_count()); + EXPECT_EQ(brpc::EPROGREADTIMEOUT, reader->end_error()); +} + +TEST_F(HttpTest, progressive_read_timeout_ignores_slow_user_callback) { + const int port = 8923; + brpc::Server server; + DownloadServiceImpl svc(DONE_BEFORE_CREATE_PA, 3, 50000); + ASSERT_EQ(0, server.AddService(&svc, brpc::SERVER_DOESNT_OWN_SERVICE)); + ASSERT_EQ(0, server.Start(port, NULL)); + + brpc::Channel channel; + brpc::ChannelOptions options; + options.protocol = brpc::PROTOCOL_HTTP; + ASSERT_EQ(0, channel.Init(butil::EndPoint(butil::my_ip(), port), &options)); + + brpc::Controller cntl; + cntl.response_will_be_read_progressively(); + cntl.set_progressive_read_timeout_ms(50); + cntl.http_request().uri() = "/DownloadService/Download"; + channel.CallMethod(NULL, &cntl, NULL, NULL, NULL); + ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText(); + + butil::intrusive_ptr reader( + new TimeoutReadBody(200000)); + cntl.ReadProgressiveAttachmentBy(reader.get()); + for (int i = 0; i < 100 && reader->end_count() == 0; ++i) { + bthread_usleep(10000); + } + ASSERT_EQ(1, reader->end_count()); + EXPECT_EQ(0, reader->end_error()); + EXPECT_EQ(3 * PA_DATA_LEN, reader->read_bytes()); +} + +TEST_F(HttpTest, progressive_read_timeout_preserves_reader_error) { + const int port = 8923; + brpc::Server server; + DownloadServiceImpl svc(DONE_BEFORE_CREATE_PA, 10); + ASSERT_EQ(0, server.AddService(&svc, brpc::SERVER_DOESNT_OWN_SERVICE)); + ASSERT_EQ(0, server.Start(port, NULL)); + + brpc::Channel channel; + brpc::ChannelOptions options; + options.protocol = brpc::PROTOCOL_HTTP; + ASSERT_EQ(0, channel.Init(butil::EndPoint(butil::my_ip(), port), &options)); + + brpc::Controller cntl; + cntl.response_will_be_read_progressively(); + cntl.set_progressive_read_timeout_ms(1000); + cntl.http_request().uri() = "/DownloadService/Download"; + channel.CallMethod(NULL, &cntl, NULL, NULL, NULL); + ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText(); + + butil::intrusive_ptr reader( + new TimeoutReadBody(0, EIO)); + cntl.ReadProgressiveAttachmentBy(reader.get()); + ASSERT_EQ(1, reader->end_count()); + EXPECT_EQ(EIO, reader->end_error()); +} + +TEST_F(HttpTest, progressive_read_timeout_rejects_http2) { + const int port = 8923; + brpc::Server server; + ASSERT_EQ(0, server.AddService(&_svc, brpc::SERVER_DOESNT_OWN_SERVICE)); + ASSERT_EQ(0, server.Start(port, NULL)); + + brpc::Channel channel; + brpc::ChannelOptions options; + options.protocol = brpc::PROTOCOL_H2; + ASSERT_EQ(0, channel.Init(butil::EndPoint(butil::my_ip(), port), &options)); + + brpc::Controller cntl; + cntl.response_will_be_read_progressively(); + cntl.set_progressive_read_timeout_ms(1000); + cntl.http_request().uri() = "/EchoService/Echo"; + test::EchoRequest req; + req.set_message(EXP_REQUEST); + channel.CallMethod(NULL, &cntl, &req, NULL, NULL); + ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText(); + + butil::intrusive_ptr reader(new TimeoutReadBody); + cntl.ReadProgressiveAttachmentBy(reader.get()); + ASSERT_EQ(1, reader->end_count()); + EXPECT_EQ(ENOTSUP, reader->end_error()); + EXPECT_EQ(size_t(0), reader->read_bytes()); +} + TEST_F(HttpTest, read_progressively_after_cntl_destroys) { DownloadServiceImpl svc(DONE_BEFORE_CREATE_PA, std::numeric_limits::max()); From a9cb054754f93fe262ecbdab9f715ecab5c21de2 Mon Sep 17 00:00:00 2001 From: zchuango Date: Fri, 7 Aug 2026 20:29:42 +0800 Subject: [PATCH 4/4] test: strengthen progressive read timeout coverage --- test/brpc_http_rpc_protocol_unittest.cpp | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/test/brpc_http_rpc_protocol_unittest.cpp b/test/brpc_http_rpc_protocol_unittest.cpp index 73e9b3dcab..6735a171b1 100644 --- a/test/brpc_http_rpc_protocol_unittest.cpp +++ b/test/brpc_http_rpc_protocol_unittest.cpp @@ -1091,7 +1091,7 @@ TEST_F(HttpTest, read_short_body_progressively) { TEST_F(HttpTest, progressive_read_timeout_keeps_active_reader_alive) { const int port = 8923; brpc::Server server; - DownloadServiceImpl svc(DONE_BEFORE_CREATE_PA, 10000); + DownloadServiceImpl svc(DONE_BEFORE_CREATE_PA, 8, 100000); ASSERT_EQ(0, server.AddService(&svc, brpc::SERVER_DOESNT_OWN_SERVICE)); ASSERT_EQ(0, server.Start(port, NULL)); @@ -1102,19 +1102,19 @@ TEST_F(HttpTest, progressive_read_timeout_keeps_active_reader_alive) { brpc::Controller cntl; cntl.response_will_be_read_progressively(); - cntl.set_progressive_read_timeout_ms(1000); + cntl.set_progressive_read_timeout_ms(500); cntl.http_request().uri() = "/DownloadService/Download"; channel.CallMethod(NULL, &cntl, NULL, NULL, NULL); ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText(); butil::intrusive_ptr reader(new TimeoutReadBody); cntl.ReadProgressiveAttachmentBy(reader.get()); - for (int i = 0; i < 100 && reader->end_count() == 0; ++i) { + for (int i = 0; i < 200 && reader->end_count() == 0; ++i) { bthread_usleep(10000); } ASSERT_EQ(1, reader->end_count()); EXPECT_EQ(0, reader->end_error()); - EXPECT_EQ(10000 * PA_DATA_LEN, reader->read_bytes()); + EXPECT_EQ(8 * PA_DATA_LEN, reader->read_bytes()); } TEST_F(HttpTest, progressive_read_timeout_closes_idle_http1_reader_once) {