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.
| 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. |
Thanks. Corresponding unit tests have been added in commit 95368db. The tests cover:
|
Centralize batch scheduling for TCP, RDMA, and UBShm transports. Make input batch cleanup exception-safe and cover batching behavior with unit tests.
95368db to
2f2dc88
Compare
|
The merge conflicts have been resolved, and this branch has been rebased onto the latest During conflict resolution, input batching was adapted to the newly introduced |
There was a problem hiding this comment.
🟡 Changes recommended
Critical and moderate issues remain, including ownership, exception-safety, and test compilation or synchronization problems.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Review details
Suppressed comments (3)
src/brpc/input_messenger.cpp:184
- This updated comment still says that all messages except the last are processed in separate bthreads or batches, but the batching path below appends
last_msgto the batch and processes it there as well. Please distinguish the unbatched last-message optimization from the batched path so the comment describes the new behavior accurately.
// - 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, or in batches when
// -input_message_batch_process_size is -1 or greater than 1. To minimize
src/brpc/input_messenger_processor.cpp:272
- The tests exercise
UpdateAdaptiveBatchSizedirectly, but none drivesProcessNewMessagewithinput_message_batch_process_size=-1. That leaves the per-socket EMA update, the real adaptive batch selection, and the reset when switching away from adaptive mode unverified; add an integration test for this branch.
const bool adaptive_batch_process =
configured_batch_size == -1 && !FLAGS_usercode_in_coroutine;
size_t batch_size = configured_batch_size > 0
? static_cast<size_t>(configured_batch_size) : 1;
if (adaptive_batch_process) {
test/brpc_input_messenger_unittest.cpp:411
- Waiting for
values.size()does not guarantee the worker has finished the handler:ProcessBatchTestMessagerecords the value before itsDestroyingPtrcallsRecordDestroy. The test then destroys the stackrecorderwithout waiting for those callbacks, so a queued batch can access a dead recorder. Wait for all four destructions before leaving the test.
ASSERT_TRUE(recorder.WaitForSize(4));
- Files reviewed: 14/14 changed files
- Comments generated: 5
- Review effort level: Lite
| InputMessageBatch() {} | ||
| explicit InputMessageBatch(size_t capacity); | ||
| ~InputMessageBatch() noexcept; |
| batch_size = brpc::InputMessengerProcessor::UpdateAdaptiveBatchSize( | ||
| &ema_q8, batch_size, std::numeric_limits<size_t>::max()); |
| socket->_transport->QueueMessages(batch, &num_bthread_created); | ||
| EXPECT_EQ((std::vector<int>{1, 2}), recorder.Snapshot()); | ||
| EXPECT_EQ(0, num_bthread_created); | ||
|
|
||
| socket->_transport->QueueMessages( |
| batch->reset(new (std::nothrow) InputMessageBatch(batch_size)); | ||
| } |
| last_msg, num_bthread_created, false); | ||
| return; | ||
| } | ||
| (*batch)->add(msg); |
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: