Copilot commented on code in PR #3428:
URL: https://github.com/apache/brpc/pull/3428#discussion_r3959030168
##########
src/brpc/input_messenger.cpp:
##########
@@ -56,294 +54,291 @@ BRPC_VALIDATE_GFLAG(log_connection_close, PassValidate);
DEFINE_bool(socket_keepalive, false,
"Enable keepalive of sockets if this value is true");
-DEFINE_int32(socket_keepalive_idle_s, -1,
- "Set idle time for socket keepalive in seconds if this value is
positive");
+DEFINE_int32(
+ socket_keepalive_idle_s, -1,
+ "Set idle time for socket keepalive in seconds if this value is positive");
-DEFINE_int32(socket_keepalive_interval_s, -1,
- "Set interval between keepalives in seconds if this value is
positive");
+DEFINE_int32(
+ socket_keepalive_interval_s, -1,
+ "Set interval between keepalives in seconds if this value is positive");
DEFINE_int32(socket_keepalive_count, -1,
"Set number of keepalives before death if this value is
positive");
-DEFINE_int32(socket_tcp_user_timeout_ms, -1,
- "If this value is positive, set number of milliseconds that
transmitted "
- "data may remain unacknowledged, or bufferred data may remain
untransmitted "
- "(due to zero window size) before TCP will forcibly close the
corresponding "
- "connection and return ETIMEDOUT to the application. Only linux
supports "
- "TCP_USER_TIMEOUT.");
+DEFINE_int32(
+ socket_tcp_user_timeout_ms, -1,
+ "If this value is positive, set number of milliseconds that transmitted "
+ "data may remain unacknowledged, or bufferred data may remain "
+ "untransmitted "
+ "(due to zero window size) before TCP will forcibly close the "
+ "corresponding "
+ "connection and return ETIMEDOUT to the application. Only linux supports "
+ "TCP_USER_TIMEOUT.");
DECLARE_bool(usercode_in_pthread);
DECLARE_bool(usercode_in_coroutine);
-void* ProcessInputMessage(void* void_arg) {
- InputMessageBase* msg = static_cast<InputMessageBase*>(void_arg);
- msg->_process(msg);
- return nullptr;
+void *ProcessInputMessage(void *void_arg) {
+ InputMessageBase *msg = static_cast<InputMessageBase *>(void_arg);
+ msg->_process(msg);
+ return nullptr;
}
struct RunLastMessage {
- inline void operator()(InputMessageBase* last_msg) {
- ProcessInputMessage(last_msg);
- }
+ inline void operator()(InputMessageBase *last_msg) {
+ ProcessInputMessage(last_msg);
+ }
};
InputMessageClosure::~InputMessageClosure() noexcept(false) {
- if (_msg) {
- ProcessInputMessage(_msg);
- }
+ if (_msg) {
+ ProcessInputMessage(_msg);
+ }
}
-void InputMessageClosure::reset(InputMessageBase* m) {
- if (_msg) {
- ProcessInputMessage(_msg);
- }
- _msg = m;
+void InputMessageClosure::reset(InputMessageBase *m) {
+ if (_msg) {
+ ProcessInputMessage(_msg);
+ }
+ _msg = m;
}
-void InputMessenger::OnNewMessages(Socket* m) {
- // Notes:
- // - If the socket has only one message, the message will be parsed and
- // processed in this bthread. nova-pbrpc and http works in this way.
- // - If the socket has several messages, all messages will be parsed (
- // meaning cutting from butil::IOBuf. serializing from protobuf is part
of
- // "process") in this bthread. All messages except the last one will be
- // processed in separate bthreads. To minimize the overhead, scheduling
- // is batched(notice the BTHREAD_NOSIGNAL and bthread_flush).
- // - Verify will always be called in this bthread at most once and before
- // any process.
- InputMessengerProcessor& processor = m->fd_input_processor();
- int progress = Socket::PROGRESS_INIT;
-
- // Notice that all *return* no matter successful or not will run last
- // message, even if the socket is about to be closed. This should be
- // OK in most cases.
- InputMessageClosure last_msg;
- bool read_eof = false;
- while (!read_eof) {
- const int64_t received_us = butil::cpuwide_time_us();
- const int64_t base_realtime = butil::gettimeofday_us() - received_us;
-
- // Read.
- const ssize_t nr = m->DoRead(&processor.read_buf(),
- processor.OnceReadSize());
- if (nr <= 0) {
- if (0 == nr) {
- // Set `read_eof' flag and proceed to feed EOF into `Protocol'
- // (implied by an empty processor.read_buf()), which may
produce
- // a new `InputMessageBase' under some protocols such as HTTP
- LOG_IF(WARNING, FLAGS_log_connection_close) << *m << " was
closed by remote side";
- read_eof = true;
- } else if (errno != EAGAIN) {
- if (errno == EINTR) {
- continue; // just retry
- }
- const int saved_errno = errno;
- PLOG(WARNING) << "Fail to read from " << *m;
- m->SetFailed(saved_errno, "Fail to read from %s: %s",
- m->description().c_str(), berror(saved_errno));
- return;
- } else if (!m->MoreReadEvents(&progress)) {
- return;
- } else { // new events during processing
- continue;
- }
- }
-
- if (processor.ProcessNewMessage(nr, read_eof, received_us,
- base_realtime, last_msg) < 0) {
- return;
+void InputMessenger::OnNewMessages(Socket *m) {
+ // Notes:
+ // - If the socket has only one message, the message will be parsed and
+ // processed in this bthread. nova-pbrpc and http works in this way.
+ // - If the socket has several messages, all messages will be parsed (
+ // meaning cutting from butil::IOBuf. serializing from protobuf is part of
+ // "process") in this bthread. All messages except the last one will be
+ // processed in separate bthreads. To minimize the overhead, scheduling
+ // is batched(notice the BTHREAD_NOSIGNAL and bthread_flush).
+ // - Verify will always be called in this bthread at most once and before
+ // any process.
+ InputMessengerProcessor &processor = m->fd_input_processor();
+ int progress = Socket::PROGRESS_INIT;
+
+ // Notice that all *return* no matter successful or not will run last
+ // message, even if the socket is about to be closed. This should be
+ // OK in most cases.
+ InputMessageClosure last_msg;
+ bool read_eof = false;
+ while (!read_eof) {
+ const int64_t received_us = butil::cpuwide_time_us();
+ const int64_t base_realtime = butil::gettimeofday_us() - received_us;
+
+ // Read.
+ const ssize_t nr =
+ m->DoRead(&processor.read_buf(), processor.OnceReadSize());
+ if (nr <= 0) {
+ if (0 == nr) {
+ // Set `read_eof' flag and proceed to feed EOF into `Protocol'
+ // (implied by an empty processor.read_buf()), which may produce
+ // a new `InputMessageBase' under some protocols such as HTTP
+ LOG_IF(WARNING, FLAGS_log_connection_close)
+ << *m << " was closed by remote side";
+ read_eof = true;
+ } else if (errno != EAGAIN) {
+ if (errno == EINTR) {
+ continue; // just retry
}
+ const int saved_errno = errno;
+ PLOG(WARNING) << "Fail to read from " << *m;
+ m->SetFailed(saved_errno, "Fail to read from %s: %s",
+ m->description().c_str(), berror(saved_errno));
+ return;
+ } else if (!m->MoreReadEvents(&progress)) {
+ return;
+ } else { // new events during processing
+ continue;
+ }
}
- if (read_eof) {
- m->SetEOF();
+ if (processor.ProcessNewMessage(nr, read_eof, received_us, base_realtime,
+ last_msg) < 0) {
+ return;
}
+ }
+
+ if (read_eof) {
+ m->SetEOF();
+ }
}
InputMessenger::InputMessenger(size_t capacity)
- : _handlers(nullptr)
- , _max_index(-1)
- , _non_protocol(false)
- , _capacity(capacity) {
-}
+ : _handlers(nullptr), _max_index(-1), _non_protocol(false),
+ _capacity(capacity) {}
InputMessenger::~InputMessenger() {
- delete[] _handlers;
- _handlers = nullptr;
- _max_index.store(-1, butil::memory_order_relaxed);
- _capacity = 0;
+ delete[] _handlers;
+ _handlers = nullptr;
+ _max_index.store(-1, butil::memory_order_relaxed);
+ _capacity = 0;
}
-int InputMessenger::AddHandler(const InputMessageHandler& handler) {
- if (handler.parse == nullptr || handler.process == nullptr
- || handler.name == nullptr) {
- CHECK(false) << "Invalid argument";
- return -1;
- }
- BAIDU_SCOPED_LOCK(_add_handler_mutex);
- if (nullptr == _handlers) {
- _handlers = new InputMessageHandler[_capacity];
- memset(_handlers, 0, sizeof(*_handlers) * _capacity);
- _non_protocol = false;
- }
- if (_non_protocol) {
- CHECK(false) << "AddNonProtocolHandler was invoked";
- return -1;
- }
- ProtocolType type = FindProtocolOfHandler(handler);
- if (type == PROTOCOL_UNKNOWN) {
- CHECK(false) << "Adding a handler which doesn't belong to any
protocol";
- return -1;
- }
- const int index = type;
- if (index >= (int)_capacity) {
- LOG(FATAL) << "Can't add more handlers than " << _capacity;
- return -1;
- }
- if (_handlers[index].parse == nullptr) {
- // The same protocol might be added more than twice
- _handlers[index] = handler;
- } else if (_handlers[index].parse != handler.parse
- || _handlers[index].process != handler.process) {
- CHECK(_handlers[index].parse == handler.parse);
- CHECK(_handlers[index].process == handler.process);
- return -1;
- }
- if (index > _max_index.load(butil::memory_order_relaxed)) {
- _max_index.store(index, butil::memory_order_release);
- }
- return 0;
-}
-
-int InputMessenger::AddNonProtocolHandler(const InputMessageHandler& handler) {
- if (handler.parse == nullptr || handler.process == nullptr
- || handler.name == nullptr) {
- CHECK(false) << "Invalid argument";
- return -1;
- }
- BAIDU_SCOPED_LOCK(_add_handler_mutex);
- if (nullptr == _handlers) {
- _handlers = new InputMessageHandler[_capacity];
- memset(_handlers, 0, sizeof(*_handlers) * _capacity);
- _non_protocol = true;
- }
- if (!_non_protocol) {
- CHECK(false) << "AddHandler was invoked";
- return -1;
- }
- const int index = _max_index.load(butil::memory_order_relaxed) + 1;
+int InputMessenger::AddHandler(const InputMessageHandler &handler) {
+ if (handler.parse == nullptr || handler.process == nullptr ||
+ handler.name == nullptr) {
+ CHECK(false) << "Invalid argument";
+ return -1;
+ }
+ BAIDU_SCOPED_LOCK(_add_handler_mutex);
+ if (nullptr == _handlers) {
+ _handlers = new InputMessageHandler[_capacity];
+ memset(_handlers, 0, sizeof(*_handlers) * _capacity);
+ _non_protocol = false;
+ }
+ if (_non_protocol) {
+ CHECK(false) << "AddNonProtocolHandler was invoked";
+ return -1;
+ }
+ ProtocolType type = FindProtocolOfHandler(handler);
+ if (type == PROTOCOL_UNKNOWN) {
+ CHECK(false) << "Adding a handler which doesn't belong to any protocol";
+ return -1;
+ }
+ const int index = type;
+ if (index >= (int)_capacity) {
+ LOG(FATAL) << "Can't add more handlers than " << _capacity;
+ return -1;
+ }
+ if (_handlers[index].parse == nullptr) {
+ // The same protocol might be added more than twice
_handlers[index] = handler;
+ } else if (_handlers[index].parse != handler.parse ||
+ _handlers[index].process != handler.process) {
+ CHECK(_handlers[index].parse == handler.parse);
+ CHECK(_handlers[index].process == handler.process);
+ return -1;
+ }
+ if (index > _max_index.load(butil::memory_order_relaxed)) {
_max_index.store(index, butil::memory_order_release);
- return 0;
+ }
+ return 0;
}
-int InputMessenger::Create(const butil::EndPoint& remote_side,
- time_t health_check_interval_s,
- SocketId* id) {
- SocketOptions options;
- options.remote_side = remote_side;
- options.user = this;
- options.on_edge_triggered_events = OnNewMessages;
- options.health_check_interval_s = health_check_interval_s;
- if (FLAGS_socket_keepalive) {
- options.keepalive_options = std::make_shared<SocketKeepaliveOptions>();
- options.keepalive_options->keepalive_idle_s
- = FLAGS_socket_keepalive_idle_s;
- options.keepalive_options->keepalive_interval_s
- = FLAGS_socket_keepalive_interval_s;
- options.keepalive_options->keepalive_count
- = FLAGS_socket_keepalive_count;
- }
- options.tcp_user_timeout_ms = FLAGS_socket_tcp_user_timeout_ms;
- return Socket::Create(options, id);
+int InputMessenger::AddNonProtocolHandler(const InputMessageHandler &handler) {
+ if (handler.parse == nullptr || handler.process == nullptr ||
+ handler.name == nullptr) {
+ CHECK(false) << "Invalid argument";
+ return -1;
+ }
+ BAIDU_SCOPED_LOCK(_add_handler_mutex);
+ if (nullptr == _handlers) {
+ _handlers = new InputMessageHandler[_capacity];
+ memset(_handlers, 0, sizeof(*_handlers) * _capacity);
+ _non_protocol = true;
+ }
+ if (!_non_protocol) {
+ CHECK(false) << "AddHandler was invoked";
+ return -1;
+ }
+ const int index = _max_index.load(butil::memory_order_relaxed) + 1;
+ _handlers[index] = handler;
+ _max_index.store(index, butil::memory_order_release);
+ return 0;
}
Review Comment:
AddNonProtocolHandler no longer bounds-checks `index` against `_capacity`,
which can cause an out-of-bounds write to `_handlers` when many non-protocol
handlers are registered (or when `_max_index` is already at capacity-1).
Restore the capacity guard (and keep behavior consistent with AddHandler)
before writing to `_handlers[index]`.
--
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]
