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.
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: