Feat: Add adaptive input message batching - #3484
Conversation
FangzuoZhang
changed the title
There was a problem hiding this comment.
Pull request overview
This PR adds an optional (experimental) input-message batching path to reduce bthread creation/scheduling overhead under bursty traffic by processing multiple parsed messages from the same socket sequentially in a single bthread, while preserving the existing behavior by default.
Changes:
- Introduces
-input_message_batch_process_size(0/1 disabled, fixed>1, adaptive-1) and implementsInputMessageBatch+ batch scheduling inInputMessenger. - Extends the internal
Transportinterface withQueueMessages(...)and implements it in TCP/RDMA/UBShm transports. - Adds per-socket adaptive batching state (EMA + current batch size) and exposes debug output for the new stats.
Reviewed changes
Copilot reviewed 11 out of 11 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| src/brpc/input_messenger.h | Adds InputMessageBatch type and batching-related helper declarations. |
| src/brpc/input_messenger.cpp | Implements batching flag, adaptive sizing logic, and batch enqueueing/flush behavior. |
| src/brpc/transport.h | Adds new pure-virtual QueueMessages API for transports. |
| src/brpc/tcp_transport.h | Declares TCP transport support for QueueMessages. |
| src/brpc/tcp_transport.cpp | Implements batch enqueue via bthread for TCP transport. |
| src/brpc/rdma_transport.h | Declares RDMA transport support for QueueMessages. |
| src/brpc/rdma_transport.cpp | Implements batch enqueue via bthread for RDMA transport. |
| src/brpc/ubshm_transport.h | Declares UBShm transport support for QueueMessages. |
| src/brpc/ubshm_transport.cpp | Implements batch enqueue via bthread for UBShm transport. |
| src/brpc/socket.h | Adds per-socket adaptive batching state fields. |
| src/brpc/socket.cpp | Initializes/resets new per-socket adaptive state; adds debug printing. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| if (!batch->get()) { | ||
| batch->reset(new (std::nothrow) InputMessageBatch(batch_size)); | ||
| } |
There was a problem hiding this comment.
The initial vector reservation is now capped at the adaptive
maximum batch size. A larger fixed batch size can still grow dynamically, but
a user-controlled flag can no longer trigger a huge upfront allocation.
| void UBShmTransport::QueueMessages(InputMessageBatch* input_msgs, | ||
| int* num_bthread_created, bool last_msg) { | ||
| CHECK(!last_msg); | ||
| if (!input_msgs || input_msgs->empty()) { | ||
| delete input_msgs; | ||
| return; | ||
| } | ||
| if (ubring::FLAGS_ub_disable_bthread) { | ||
| input_msgs->Run(); | ||
| delete input_msgs; | ||
| return; | ||
| } |
There was a problem hiding this comment.
Removed the unused last_msg parameter from the batch scheduling interface.
The final RDMA/UBRing message continues to use the single-message path.
| void RdmaTransport::QueueMessages(InputMessageBatch* input_msgs, | ||
| int* num_bthread_created, bool last_msg) { | ||
| CHECK(!last_msg || rdma::FLAGS_rdma_use_polling); | ||
| if (!input_msgs || input_msgs->empty()) { | ||
| delete input_msgs; | ||
| return; | ||
| } | ||
| if (rdma::FLAGS_rdma_disable_bthread) { | ||
| input_msgs->Run(); | ||
| delete input_msgs; | ||
| return; | ||
| } |
There was a problem hiding this comment.
Removed the unused last_msg parameter from the batch scheduling interface.
The final RDMA/UBRing message continues to use the single-message path.
| public: | ||
| InputMessageBatch() {} | ||
| explicit InputMessageBatch(size_t capacity); | ||
| ~InputMessageBatch() noexcept(false); |
There was a problem hiding this comment.
Addressed in commit 95368db.
InputMessageBatch::~InputMessageBatch is now noexcept. Before invoking a handler, Run() clears the corresponding message slot so that an already-delivered message cannot be processed again.
The destructor catches all exceptions raised during handler execution, logs an English error message, and destroys any remaining undelivered messages. The bthread batch entry also contains handler exceptions so that they cannot escape through the bthread callback boundary.
The unit test verifies the nothrow destructor property, injects a handler exception, confirms that the exception does not propagate, and checks that the remaining messages are cleaned up exactly once.
| 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); | ||
| } |
There was a problem hiding this comment.
Addressed in commit 95368db.
When batching is enabled, the final last_msg is now appended to the current InputMessageBatch before the batch is scheduled. It is therefore processed sequentially with the preceding messages in that batch instead of being scheduled separately.
When batching is disabled, the existing RDMA/UBRing last-message optimization is preserved. Progressive-read messages also remain on the original individual-message path.
The batching_consumes_the_last_message unit test verifies that the final message is consumed by the current batch, processed in parsing order, and no longer remains in InputMessageClosure.
| 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; | ||
| } | ||
| } |
There was a problem hiding this comment.
Addressed in commit 95368db.
I extracted the common batch scheduling logic into Transport::QueueInputMessageBatch. The shared helper now handles:
- empty batches;
- transport-specific synchronous execution;
- bthread attributes, name, keytable pool, and tag;
- synchronous fallback when bthread creation fails;
num_bthread_createdaccounting.
TcpTransport, RdmaTransport, and UBShmTransport now only provide their transport-specific synchronous execution condition and delegate the remaining work to the shared helper.
The unit tests cover the empty-batch path, synchronous execution, asynchronous scheduling, and bthread creation accounting.
|
Please also add the corresponding unit tests. |
Centralize batch scheduling for TCP, RDMA, and UBShm transports. Make input batch cleanup exception-safe and cover batching behavior with unit tests.
Thanks. Corresponding unit tests have been added in commit 95368db. The tests cover:
|
What problem does this PR solve?
Issue Number: N/A
Related work:
Problem Summary:
InputMessengercurrently schedules each parsed input message in a separatebthread. For short message handlers and bursty traffic on a single connection,
bthread creation and scheduling may account for a significant portion of the
request-processing overhead.
This PR introduces optional input message batching. Messages parsed from the
same socket can be processed sequentially in one bthread, reducing scheduling
overhead while preserving the existing behavior by default.
What is changed and the side effects?
Design
Changed:
input_message_batch_process_sizegflag:0or1: preserve the original one-message-per-bthread behavior.1: use a fixed batch size.-1: adaptively select a batch size from1,2,4,8, and16.-1are rejected.InputMessageBatchto own and process messages in their original order.average of messages parsed from each read.
the observed burst size drops.
fails.
Side effects:
Performance effects:
The default value is
0, so existing deployments retain the originalscheduling behavior.
When batching is enabled, it reduces bthread creation and scheduling overhead
for bursty workloads. A larger fixed batch may increase the time that later
messages wait behind earlier handlers. Adaptive mode limits the maximum batch
size to 16 and decreases the batch size quickly when the observed burst size
drops.
Breaking backward compatibility:
There is no change to the public RPC protocol or default runtime behavior.
The internal
Transportinterface gains aQueueMessagesvirtual method.Downstream custom transport implementations derived directly from
Transportmust implement this method.Performance test
The RDMA performance example was used with a single connection and multiple
outstanding requests on that connection. Each attachment size was tested three
times.
input_message_batch_process_size=0input_message_batch_process_size=-1queue_depthmust be greater than 1 to produce message bursts on the sameconnection.
usage.
baseline.
Average results
The following results were measured on a Kunpeng 950 server.
Each attachment size was tested three times, and the reported values are the
arithmetic averages of those runs.
input_message_batch_process_size=0input_message_batch_process_size=-1usage.
Check List: