Copilot commented on code in PR #3484:
URL: https://github.com/apache/brpc/pull/3484#discussion_r3871873105
##########
src/brpc/input_messenger.cpp:
##########
@@ -301,9 +454,20 @@ int InputMessenger::ProcessNewMessage(
// 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 (m->_socket_mode == SOCKET_MODE_RDMA || m->_socket_mode ==
SOCKET_MODE_UBRING) {
+ if (batch_process) {
+ QueueInputMessageBatch(m, &input_batch, &num_bthread_created);
+ }
+ if (m->_socket_mode == SOCKET_MODE_RDMA ||
+ m->_socket_mode == SOCKET_MODE_UBRING) {
m->_transport->QueueMessage(last_msg, &num_bthread_created, true);
}
Review Comment:
When `batch_process` is enabled, earlier messages are queued as one batch
(potentially in a dedicated bthread), but the last message is still queued
separately (for RDMA/UBRING here). This means processing for a single read may
still span multiple bthreads, which conflicts with the stated design goal of
processing messages from the same socket sequentially in one bthread, and it
can also allow the 'last message' to run concurrently/out-of-order relative to
the batch. If sequential-in-one-bthread is an intended guarantee under
batching, consider appending `last_msg` into the same `InputMessageBatch` when
batching is active (and use one scheduling call), or otherwise enforce ordering
between the batch and the final message.
##########
src/brpc/tcp_transport.cpp:
##########
@@ -103,4 +103,25 @@ void TcpTransport::QueueMessage(InputMessageClosure&
input_msg,
}
}
+void TcpTransport::QueueMessages(InputMessageBatch* input_msgs,
+ int* num_bthread_created) {
+ if (!input_msgs || input_msgs->empty()) {
+ delete input_msgs;
+ return;
+ }
+ bthread_t th;
+ bthread_attr_t tmp =
+ (FLAGS_usercode_in_pthread ? BTHREAD_ATTR_PTHREAD :
BTHREAD_ATTR_NORMAL) |
+ BTHREAD_NOSIGNAL;
+ tmp.keytable_pool = _socket->keytable_pool();
+ tmp.tag = bthread_self_tag();
+ if (!FLAGS_usercode_in_coroutine && bthread_start_background(
+ &th, &tmp, ProcessInputMessageBatch, input_msgs) == 0) {
+ ++*num_bthread_created;
+ } else {
+ input_msgs->Run();
+ delete input_msgs;
+ }
+}
Review Comment:
`QueueMessages(...)` is implemented with largely duplicated logic across
`TcpTransport`, `RdmaTransport`, and `UBShmTransport` (setup attrs, start
bthread, fallback to synchronous + delete). This duplication increases the
chance of drift (e.g., UBShm sets a bthread name but TCP/RDMA currently do
not). Consider extracting a shared helper (e.g., a small internal utility
function) to centralize the common scheduling/fallback behavior and keep
attributes/logging consistent across transports.
##########
src/brpc/input_messenger.h:
##########
@@ -91,6 +94,24 @@ class InputMessageClosure {
InputMessageBase* _msg;
};
+class InputMessageBatch {
+public:
+ InputMessageBatch() {}
+ explicit InputMessageBatch(size_t capacity);
+ ~InputMessageBatch() noexcept(false);
Review Comment:
Declaring the destructor as `noexcept(false)` is risky here because
`InputMessageBatch` destruction can happen on error paths and in bthread entry
cleanup, and `Run()` ultimately invokes user handlers (via
`ProcessInputMessage`) that may throw if exceptions are enabled. Letting
exceptions propagate out of a destructor is unsafe (can trigger
`std::terminate` during stack unwinding) and also makes exception guarantees
harder to reason about. Prefer making the destructor `noexcept` and ensuring it
cannot surface exceptions (e.g., avoid calling `Run()` from the destructor, or
wrap `Run()` in a catch-all that logs and clears remaining messages).
--
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]
