Copilot commented on code in PR #3409:
URL: https://github.com/apache/brpc/pull/3409#discussion_r3679786422
##########
src/brpc/progressive_reader.h:
##########
@@ -20,6 +20,7 @@
#define BRPC_PROGRESSIVE_READER_H
#include "brpc/shared_object.h"
+#include "brpc/socket.h"
Review Comment:
Including brpc/socket.h here is unnecessarily heavy for this header;
ReadableProgressiveAttachment only needs the SocketId type. Prefer including
brpc/socket_id.h to avoid pulling in the full Socket API (reduces compile time
and header coupling).
##########
src/brpc/controller.cpp:
##########
@@ -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);
+ }
Review Comment:
This timeout reader is intended to detect *idle* progressive reads, but the
timer is only armed once in the constructor. As written, it triggers after a
fixed duration from the start even if data is continuously arriving. Re-arm the
timer whenever a new part is read.
This issue also appears on line 199 of the same file.
##########
src/brpc/controller.cpp:
##########
@@ -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<ProgressiveTimeoutReader*>(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);
Review Comment:
Destroying s->parsing_context() from a timer thread is unsafe:
http_rpc_protocol.cpp explicitly notes that calling HttpContext::Destroy()
directly is wrong outside ProcessHttpXXX, and this also risks races with the
socket's parsing thread. To close the connection on timeout, mark the socket
failed instead of destroying the parsing context here.
##########
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");
Review Comment:
Spelling: "attachement" -> "attachment" in the flag description (also makes
the help output more professional).
##########
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
Review Comment:
EPROGREADTIMEOUT is added to errno.proto, but there is no corresponding
BAIDU_REGISTER_ERRNO mapping in controller.cpp. Without registering, code paths
that rely on the errno registry may show an unhelpful numeric code. Please
register this errno alongside the other client-side errnos.
##########
src/brpc/controller.cpp:
##########
@@ -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";
+ }
+}
Review Comment:
This clamp/check is ineffective because the parameter type is int32_t (it
can never exceed 0x7fffffff). The warning message also mentions "seconds" even
though this API is in milliseconds. Consider simplifying the setter to a
straight assignment (or change the API to accept int64_t if you really need
clamping).
##########
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
Review Comment:
HttpContext initializes _socket_id to 0, but the codebase defines
INVALID_SOCKET_ID for the sentinel value. Using 0 can mask bugs in cases where
SetSocketId() was not called (e.g., UTs), and it may result in failed lookups
against Socket::Address().
##########
src/brpc/controller.cpp:
##########
@@ -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<ProgressiveTimeoutReader*>(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;
Review Comment:
The log message says the timeout is in "us", but _read_timeout_ms is in
milliseconds and is used with milliseconds_from_now(). This makes the log
misleading when diagnosing timeouts.
##########
example/http_c++/http_client.cpp:
##########
@@ -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();
+ }
Review Comment:
This example reader copies arbitrary-length data into a fixed 1KB buffer
without bounds checking and then logs it as a C string without ensuring
NUL-termination. This can overflow the buffer and/or read past the copied data
when logging.
This issue also appears on line 52 of the same file.
##########
src/brpc/controller.cpp:
##########
@@ -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);
+ }
Review Comment:
New progressive-read timeout behavior is introduced here, but there are
existing progressive-read unit tests (test/brpc_http_rpc_protocol_unittest.cpp)
and none cover the timeout path. Please add a unit test that verifies an idle
progressive read closes the socket and the reader observes EPROGREADTIMEOUT (or
the expected failure).
--
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]
