Copilot commented on code in PR #3428:
URL: https://github.com/apache/brpc/pull/3428#discussion_r3960505065
##########
src/brpc/input_messenger_processor.cpp:
##########
@@ -15,279 +15,285 @@
// specific language governing permissions and limitations
// under the License.
-#include "butil/logging.h"
-#include "butil/binary_printer.h"
-#include "bthread/unstable.h"
-#include "brpc/options.pb.h"
-#include "brpc/transport.h"
#include "brpc/input_messenger_processor.h"
#include "brpc/input_messenger.h"
+#include "brpc/options.pb.h"
+#include "brpc/transport.h"
+#include "bthread/unstable.h"
+#include "butil/binary_printer.h"
+#include "butil/logging.h"
+
namespace brpc {
DECLARE_uint64(max_body_size);
-const size_t MSG_SIZE_WINDOW = 10; // Take last so many message into stat.
+const size_t MSG_SIZE_WINDOW = 10; // Take last so many message into stat.
const size_t MIN_ONCE_READ = 4096;
const size_t MAX_ONCE_READ = 524288;
-static const char* StreamTypeName(InputMessengerProcessor::StreamType type) {
- switch (type) {
- case InputMessengerProcessor::STREAM_NONE: return "none";
- case InputMessengerProcessor::STREAM_TCP_FD: return "tcp_fd";
- case InputMessengerProcessor::STREAM_RDMA_QP: return "rdma_qp";
- }
- return "unknown";
+static const char *StreamTypeName(InputMessengerProcessor::StreamType type) {
+ switch (type) {
+ case InputMessengerProcessor::STREAM_NONE:
+ return "none";
+ case InputMessengerProcessor::STREAM_TCP_FD:
+ return "tcp_fd";
+ case InputMessengerProcessor::STREAM_RDMA_QP:
+ return "rdma_qp";
+ case InputMessengerProcessor::STREAM_URMA_JETTY:
+ return "urma_jetty";
+ }
+ return "unknown";
}
InputMessengerProcessor::ParsingStreamGuard::ParsingStreamGuard(Socket*
socket, StreamType type)
: _socket(socket) {
- CHECK(socket != nullptr)
- << "Parsing through a processor that was never Init()ed, socket is
NULL";
- CHECK_NE(STREAM_NONE, type)
- << "Parsing through a processor that was never Init()ed, " << *socket;
- CHECK_EQ(STREAM_NONE, socket->parsing_stream_type())
- << "Two input streams of " << *socket << " are parsing at the same
time: "
- << StreamTypeName(socket->parsing_stream_type()) << " and "
- << StreamTypeName(type);
- socket->set_parsing_stream_type(type);
+ CHECK(socket != nullptr)
+ << "Parsing through a processor that was never Init()ed, socket is NULL";
+ CHECK_NE(STREAM_NONE, type)
+ << "Parsing through a processor that was never Init()ed, " << *socket;
+ CHECK_EQ(STREAM_NONE, socket->parsing_stream_type())
+ << "Two input streams of " << *socket << " are parsing at the same time:
"
+ << StreamTypeName(socket->parsing_stream_type()) << " and "
+ << StreamTypeName(type);
+ socket->set_parsing_stream_type(type);
}
InputMessengerProcessor::ParsingStreamGuard::~ParsingStreamGuard() {
- _socket->set_parsing_stream_type(STREAM_NONE);
+ _socket->set_parsing_stream_type(STREAM_NONE);
}
-ParseResult InputMessengerProcessor::CutInputMessage(InputMessenger* messenger,
- size_t* index, bool
read_eof) {
- ParsingStreamGuard parsing_stream_guard(_socket, _stream_type);
- InputMessageHandler* handlers = messenger->_handlers;
- int preferred = _socket->preferred_index();
- int max_index =
(int)messenger->_max_index.load(butil::memory_order_acquire);
- // Try preferred handler first. The preferred_index is set on last
- // selection or by client.
- if (preferred >= 0 && preferred <= max_index
- && handlers[preferred].parse != nullptr) {
- int cur_index = preferred;
- do {
- ParseResult result =
- handlers[cur_index].parse(&_read_buf, _socket, read_eof,
- handlers[cur_index].arg);
- if (result.is_ok() ||
- result.error() == PARSE_ERROR_NOT_ENOUGH_DATA) {
- _socket->set_preferred_index(cur_index);
- *index = cur_index;
- return result;
- } else if (result.error() != PARSE_ERROR_TRY_OTHERS) {
- // Critical error, return directly.
- LOG_IF(ERROR, result.error() == PARSE_ERROR_TOO_BIG_DATA)
- << "A message from " << _socket->remote_side()
- << "(protocol=" << handlers[cur_index].name
- << ") is bigger than " << FLAGS_max_body_size
- << " bytes, the connection will be closed."
- " Set max_body_size to allow bigger messages";
- return result;
- }
+ParseResult InputMessengerProcessor::CutInputMessage(InputMessenger *messenger,
+ size_t *index,
+ bool read_eof) {
+ ParsingStreamGuard parsing_stream_guard(_socket, _stream_type);
+ InputMessageHandler *handlers = messenger->_handlers;
+ int preferred = _socket->preferred_index();
+ int max_index = (int)messenger->_max_index.load(butil::memory_order_acquire);
+ // Try preferred handler first. The preferred_index is set on last
+ // selection or by client.
+ if (preferred >= 0 && preferred <= max_index &&
+ handlers[preferred].parse != nullptr) {
+ int cur_index = preferred;
+ do {
+ ParseResult result = handlers[cur_index].parse(
+ &_read_buf, _socket, read_eof, handlers[cur_index].arg);
+ if (result.is_ok() || result.error() == PARSE_ERROR_NOT_ENOUGH_DATA) {
+ _socket->set_preferred_index(cur_index);
+ *index = cur_index;
+ return result;
+ } else if (result.error() != PARSE_ERROR_TRY_OTHERS) {
+ // Critical error, return directly.
+ LOG_IF(ERROR, result.error() == PARSE_ERROR_TOO_BIG_DATA)
+ << "A message from " << _socket->remote_side()
+ << "(protocol=" << handlers[cur_index].name << ") is bigger than "
+ << FLAGS_max_body_size
+ << " bytes, the connection will be closed."
+ " Set max_body_size to allow bigger messages";
+ return result;
+ }
- if (_socket->CreatedByConnect()) {
- if((ProtocolType)cur_index == PROTOCOL_BAIDU_STD && cur_index
== preferred) {
- // baidu_std may fall to streaming_rpc.
- cur_index = (int)PROTOCOL_STREAMING_RPC;
- continue;
- } else if((ProtocolType)cur_index == PROTOCOL_STREAMING_RPC &&
- cur_index == preferred) {
- // streaming_rpc may fall to baidu_std.
- cur_index = (int)PROTOCOL_BAIDU_STD;
- continue;
- } else {
- // The protocol is fixed at client-side, no need to try
others.
- LOG(ERROR) << "Fail to parse response from " <<
_socket->remote_side()
- << " by " << handlers[preferred].name
- << " at client-side";
- return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
- }
- } else {
- // Try other protocols.
- //
- // A handler may lean on this: returning
PARSE_ERROR_NOT_ENOUGH_DATA
- // above keeps `preferred_index' pinned on it, TRY_OTHERS here
gives
- // it up. RdmaEndpoint::ExecuteServerHandshake() pins itself
that way
- // to keep the last handshake read from being taken for a
protocol
- // detection. See the tail of its phase 2 before changing what
- // happens to `preferred_index' here.
- break;
- }
- } while (true);
- // Clear context before trying next protocol which probably has
- // an incompatible context with the current one.
- if (_socket->parsing_context()) {
- _socket->reset_parsing_context(nullptr);
- }
- _socket->set_preferred_index(-1);
- }
- for (int i = 0; i <= max_index; ++i) {
- if (i == preferred || handlers[i].parse == nullptr) {
- // Don't try preferred handler(already tried) or invalid handler
- continue;
- }
- ParseResult result = handlers[i].parse(&_read_buf, _socket, read_eof,
handlers[i].arg);
- if (result.is_ok() ||
- result.error() == PARSE_ERROR_NOT_ENOUGH_DATA) {
- _socket->set_preferred_index(i);
- *index = i;
- return result;
- } else if (result.error() != PARSE_ERROR_TRY_OTHERS) {
- // Critical error, return directly.
- LOG_IF(ERROR, result.error() == PARSE_ERROR_TOO_BIG_DATA)
- << "A message from " << _socket->remote_side()
- << "(protocol=" << handlers[i].name
- << ") is bigger than " << FLAGS_max_body_size
- << " bytes, the connection will be closed."
- " Set max_body_size to allow bigger messages";
- return result;
- }
- // Clear context before trying next protocol which definitely has
- // an incompatible context with the current one.
- if (_socket->parsing_context()) {
- _socket->reset_parsing_context(nullptr);
+ if (_socket->CreatedByConnect()) {
+ if ((ProtocolType)cur_index == PROTOCOL_BAIDU_STD &&
+ cur_index == preferred) {
+ // baidu_std may fall to streaming_rpc.
+ cur_index = (int)PROTOCOL_STREAMING_RPC;
+ continue;
+ } else if ((ProtocolType)cur_index == PROTOCOL_STREAMING_RPC &&
+ cur_index == preferred) {
+ // streaming_rpc may fall to baidu_std.
+ cur_index = (int)PROTOCOL_BAIDU_STD;
+ continue;
+ } else {
+ // The protocol is fixed at client-side, no need to try others.
+ LOG(ERROR) << "Fail to parse response from " <<
_socket->remote_side()
+ << " by " << handlers[preferred].name << " at
client-side";
+ return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
}
+ } else {
// Try other protocols.
+ //
+ // A handler may lean on this: returning PARSE_ERROR_NOT_ENOUGH_DATA
+ // above keeps `preferred_index' pinned on it, TRY_OTHERS here gives
+ // it up. RdmaEndpoint::ExecuteServerHandshake() pins itself that way
+ // to keep the last handshake read from being taken for a protocol
+ // detection. See the tail of its phase 2 before changing what
+ // happens to `preferred_index' here.
+ break;
+ }
+ } while (true);
+ // Clear context before trying next protocol which probably has
+ // an incompatible context with the current one.
+ if (_socket->parsing_context()) {
+ _socket->reset_parsing_context(nullptr);
+ }
+ _socket->set_preferred_index(-1);
+ }
+ for (int i = 0; i <= max_index; ++i) {
+ if (i == preferred || handlers[i].parse == nullptr) {
+ // Don't try preferred handler(already tried) or invalid handler
+ continue;
}
- return MakeParseError(PARSE_ERROR_TRY_OTHERS);
+ ParseResult result =
+ handlers[i].parse(&_read_buf, _socket, read_eof, handlers[i].arg);
+ if (result.is_ok() || result.error() == PARSE_ERROR_NOT_ENOUGH_DATA) {
+ _socket->set_preferred_index(i);
+ *index = i;
+ return result;
+ } else if (result.error() != PARSE_ERROR_TRY_OTHERS) {
+ // Critical error, return directly.
+ LOG_IF(ERROR, result.error() == PARSE_ERROR_TOO_BIG_DATA)
+ << "A message from " << _socket->remote_side()
+ << "(protocol=" << handlers[i].name << ") is bigger than "
+ << FLAGS_max_body_size
+ << " bytes, the connection will be closed."
+ " Set max_body_size to allow bigger messages";
+ return result;
+ }
+ // Clear context before trying next protocol which definitely has
+ // an incompatible context with the current one.
+ if (_socket->parsing_context()) {
+ _socket->reset_parsing_context(nullptr);
+ }
+ // Try other protocols.
+ }
+ return MakeParseError(PARSE_ERROR_TRY_OTHERS);
}
size_t InputMessengerProcessor::OnceReadSize() const {
- size_t once_read = _avg_msg_size * 16;
- if (once_read < MIN_ONCE_READ) {
- once_read = MIN_ONCE_READ;
- } else if (once_read > MAX_ONCE_READ) {
- once_read = MAX_ONCE_READ;
- }
- return once_read;
+ size_t once_read = _avg_msg_size * 16;
+ if (once_read < MIN_ONCE_READ) {
+ once_read = MIN_ONCE_READ;
+ } else if (once_read > MAX_ONCE_READ) {
+ once_read = MAX_ONCE_READ;
+ }
+ return once_read;
}
void InputMessengerProcessor::Reset() {
- _read_buf.clear();
- _last_msg_size = 0;
- _avg_msg_size = 0;
+ _read_buf.clear();
+ _last_msg_size = 0;
+ _avg_msg_size = 0;
}
int InputMessengerProcessor::ProcessNewMessage(ssize_t bytes, bool read_eof,
uint64_t received_us,
uint64_t base_realtime,
- InputMessageClosure& last_msg) {
- auto messenger = static_cast<InputMessenger*>(_socket->user());
- const InputMessageHandler* handlers = messenger->_handlers;
- _socket->AddInputBytes(bytes);
+ InputMessageClosure &last_msg) {
+ auto messenger = static_cast<InputMessenger *>(_socket->user());
+ const InputMessageHandler *handlers = messenger->_handlers;
+ _socket->AddInputBytes(bytes);
- // Avoid this socket to be closed due to idle_timeout_s
- _socket->_last_readtime_us.store(received_us, butil::memory_order_relaxed);
+ // Avoid this socket to be closed due to idle_timeout_s
+ _socket->_last_readtime_us.store(received_us, butil::memory_order_relaxed);
- size_t last_size = _read_buf.length();
- int num_bthread_created = 0;
- while (true) {
- size_t index = 8888;
- ParseResult pr = CutInputMessage(messenger, &index, read_eof);
- if (!pr.is_ok()) {
- if (pr.error() == PARSE_ERROR_NOT_ENOUGH_DATA) {
- // incomplete message, re-read.
- // However, some buffer may have been consumed
- // under protocols like HTTP. Record this size
- _last_msg_size += (last_size - _read_buf.length());
- break;
- } else if (pr.error() == PARSE_ERROR_TRY_OTHERS) {
- LOG(WARNING) << "Close " << *_socket << " due to unknown
message: "
- << butil::ToPrintable(_read_buf);
- _socket->SetFailed(EINVAL, "Close %s due to unknown message",
- _socket->description().c_str());
- return -1;
- } else {
- LOG(WARNING) << "Close " << *_socket << ": " << pr.error_str();
- _socket->SetFailed(EINVAL, "Close %s: %s",
- _socket->description().c_str(),
pr.error_str());
- return -1;
- }
- }
+ size_t last_size = _read_buf.length();
+ int num_bthread_created = 0;
+ while (true) {
+ size_t index = 8888;
+ ParseResult pr = CutInputMessage(messenger, &index, read_eof);
+ if (!pr.is_ok()) {
+ if (pr.error() == PARSE_ERROR_NOT_ENOUGH_DATA) {
+ // incomplete message, re-read.
+ // However, some buffer may have been consumed
+ // under protocols like HTTP. Record this size
+ _last_msg_size += (last_size - _read_buf.length());
+ break;
+ } else if (pr.error() == PARSE_ERROR_TRY_OTHERS) {
+ LOG(WARNING) << "Close " << *_socket << " due to unknown message: "
+ << butil::ToPrintable(_read_buf);
+ _socket->SetFailed(EINVAL, "Close %s due to unknown message",
+ _socket->description().c_str());
+ return -1;
+ } else {
+ LOG(WARNING) << "Close " << *_socket << ": " << pr.error_str();
+ _socket->SetFailed(EINVAL, "Close %s: %s",
+ _socket->description().c_str(), pr.error_str());
+ return -1;
+ }
+ }
- _socket->AddInputMessages(1);
- // Calculate average size of messages
- const size_t cur_size = _read_buf.length();
- if (cur_size == 0) {
- // _read_buf is consumed, it's good timing to return blocks
- // cached internally back to TLS, otherwise the memory is not
- // reused until next message arrives which is quite uncertain
- // in situations that most connections are idle.
- _read_buf.return_cached_blocks();
- }
- _last_msg_size += (last_size - cur_size);
- last_size = cur_size;
- const size_t old_avg = _avg_msg_size;
- if (old_avg != 0) {
- _avg_msg_size = (old_avg * (MSG_SIZE_WINDOW - 1) + _last_msg_size)
/ MSG_SIZE_WINDOW;
- } else {
- _avg_msg_size = _last_msg_size;
- }
- _last_msg_size = 0;
+ _socket->AddInputMessages(1);
+ // Calculate average size of messages
+ const size_t cur_size = _read_buf.length();
+ if (cur_size == 0) {
+ // _read_buf is consumed, it's good timing to return blocks
+ // cached internally back to TLS, otherwise the memory is not
+ // reused until next message arrives which is quite uncertain
+ // in situations that most connections are idle.
+ _read_buf.return_cached_blocks();
+ }
+ _last_msg_size += (last_size - cur_size);
+ last_size = cur_size;
+ const size_t old_avg = _avg_msg_size;
+ if (old_avg != 0) {
+ _avg_msg_size =
+ (old_avg * (MSG_SIZE_WINDOW - 1) + _last_msg_size) / MSG_SIZE_WINDOW;
+ } else {
+ _avg_msg_size = _last_msg_size;
+ }
+ _last_msg_size = 0;
- if (pr.message() == nullptr) { // the Process() step can be skipped.
- continue;
- }
- pr.message()->_received_us = received_us;
- pr.message()->_base_real_us = base_realtime;
+ if (pr.message() == nullptr) { // the Process() step can be skipped.
+ continue;
+ }
+ pr.message()->_received_us = received_us;
+ pr.message()->_base_real_us = base_realtime;
- // This unique_ptr prevents msg to be lost before transfering
- // ownership to last_msg
- DestroyingPtr<InputMessageBase> msg(pr.message());
- _socket->_transport->QueueMessage(last_msg, &num_bthread_created,
false);
- if (handlers[index].process == nullptr) {
- LOG(ERROR) << "process of index=" << index << " is NULL";
- continue;
- }
- _socket->ReAddress(&msg->_socket);
- _socket->PostponeEOF();
- msg->_process = handlers[index].process;
- msg->_arg = handlers[index].arg;
+ // This unique_ptr prevents msg to be lost before transfering
+ // ownership to last_msg
+ DestroyingPtr<InputMessageBase> msg(pr.message());
+ _socket->_transport->QueueMessage(last_msg, &num_bthread_created, false);
+ if (handlers[index].process == nullptr) {
+ LOG(ERROR) << "process of index=" << index << " is NULL";
+ continue;
+ }
+ _socket->ReAddress(&msg->_socket);
+ _socket->PostponeEOF();
+ msg->_process = handlers[index].process;
+ msg->_arg = handlers[index].arg;
- if (handlers[index].verify != nullptr) {
- int auth_error = 0;
- if (0 == _socket->FightAuthentication(&auth_error)) {
- // Get the right to authenticate
- if (handlers[index].verify(msg.get())) {
- _socket->SetAuthentication(0);
- } else {
- _socket->SetAuthentication(ERPCAUTH);
- LOG(WARNING) << "Fail to authenticate " << *_socket;
- _socket->SetFailed(ERPCAUTH, "Fail to authenticate %s",
- _socket->description().c_str());
- return -1;
- }
- } else {
- LOG_IF(FATAL, auth_error != 0) <<
- "Impossible! Socket should have been "
- "destroyed when authentication failed";
- }
- }
- if (!_socket->is_read_progressive()) {
- // Transfer ownership to last_msg
- last_msg.reset(msg.release());
+ if (handlers[index].verify != nullptr) {
+ int auth_error = 0;
+ if (0 == _socket->FightAuthentication(&auth_error)) {
+ // Get the right to authenticate
+ if (handlers[index].verify(msg.get())) {
+ _socket->SetAuthentication(0);
} else {
- last_msg.reset(msg.release());
- _socket->_transport->QueueMessage(last_msg, &num_bthread_created,
false);
- bthread_flush();
- num_bthread_created = 0;
+ _socket->SetAuthentication(ERPCAUTH);
+ LOG(WARNING) << "Fail to authenticate " << *_socket;
+ _socket->SetFailed(ERPCAUTH, "Fail to authenticate %s",
+ _socket->description().c_str());
+ return -1;
}
+ } else {
+ LOG_IF(FATAL, auth_error != 0)
+ << "Impossible! Socket should have been "
+ "destroyed when authentication failed";
+ }
}
- // In RDMA polling mode, all messages must be executed in a new bthread and
- // not in the bthread where the polling bthread is located, because the
- // method for processing messages may call synchronization primitives,
- // causing the polling bthread to be scheduled out.
- if (_socket->_socket_mode == SOCKET_MODE_RDMA ||
- _socket->_socket_mode == SOCKET_MODE_UBRING) {
- _socket->_transport->QueueMessage(last_msg, &num_bthread_created,
true);
- }
- if (num_bthread_created) {
- bthread_flush();
+ if (!_socket->is_read_progressive()) {
+ // Transfer ownership to last_msg
+ last_msg.reset(msg.release());
+ } else {
+ last_msg.reset(msg.release());
+ _socket->_transport->QueueMessage(last_msg, &num_bthread_created, false);
+ bthread_flush();
+ num_bthread_created = 0;
}
- return 0;
+ }
+ // On dedicated transport pollers (RDMA, UBRING, and URMA), all messages must
+ // be executed in a new bthread. Processing user code on the poller may call
+ // synchronization primitives and prevent it from draining more events.
+ if (_socket->_socket_mode == SOCKET_MODE_RDMA ||
+ _socket->_socket_mode == SOCKET_MODE_UBRING ||
+ _socket->_socket_mode == SOCKET_MODE_URMA) {
Review Comment:
This comment states that URMA always behaves like a dedicated poller that
must not run user code inline, but `UrmaTransport::QueueMessage` currently
returns early when `--urma_use_polling=false`, causing the last message to run
inline via `InputMessageClosure` destruction. Either narrow the
comment/condition to URMA polling mode only, or adjust URMA’s queuing behavior
so it matches the stated constraint.
##########
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 writes `_handlers[index]` without checking `index <
_capacity`, which can cause out-of-bounds writes when multiple non-protocol
handlers are added. Add the same capacity guard used in `AddHandler` (e.g.,
fatal/return error when `index >= (int)_capacity`).
--
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]
