Copilot commented on code in PR #3484:
URL: https://github.com/apache/brpc/pull/3484#discussion_r3844508256
##########
src/brpc/input_messenger.cpp:
##########
@@ -192,6 +233,79 @@ void InputMessageClosure::reset(InputMessageBase* m) {
_msg = m;
}
+void InputMessenger::QueueInputMessageBatch(
+ Socket* m, std::unique_ptr<InputMessageBatch>* batch,
+ int* num_bthread_created, bool last_msg) {
+ if (!batch->get() || (*batch)->empty()) {
+ return;
+ }
+ m->_transport->QueueMessages(
+ batch->release(), num_bthread_created, last_msg);
+}
+
+void InputMessenger::QueueLastMessageOrBatch(
+ Socket* m, InputMessageClosure& last_msg,
+ std::unique_ptr<InputMessageBatch>* batch,
+ int* num_bthread_created, size_t batch_size) {
+ InputMessageBase* msg = last_msg.release();
+ if (!msg) {
+ return;
+ }
+ if (!batch->get()) {
+ batch->reset(new (std::nothrow) InputMessageBatch(batch_size));
+ }
Review Comment:
`InputMessageBatch(batch_size)` reserves `batch_size` elements, which is
user-controlled via `-input_message_batch_process_size`. A very large value can
cause a large allocation (or termination on OOM) even though `new
(std::nothrow)` is used for fallback. Consider reserving at most a small
bounded capacity (e.g. the adaptive max) to avoid huge upfront allocations
while still allowing large fixed batch sizes if desired.
##########
src/brpc/ubshm_transport.cpp:
##########
@@ -174,6 +174,35 @@ void UBShmTransport::QueueMessage(InputMessageClosure&
input_msg,
}
}
+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;
+ }
Review Comment:
`QueueMessages` uses `CHECK(!last_msg)`, which will abort the process if
`last_msg` is ever passed as true (e.g. due to a future call-site change or
bug). Since this is an internal scheduling hint, it’s safer to handle the case
gracefully (run inline and return) rather than crashing.
##########
src/brpc/rdma_transport.cpp:
##########
@@ -183,6 +183,34 @@ void RdmaTransport::QueueMessage(InputMessageClosure&
input_msg,
}
}
+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;
+ }
Review Comment:
`QueueMessages` uses a `CHECK(!last_msg || rdma_use_polling)` guard. A wrong
`last_msg` value would currently crash the process; it’s safer to handle the
case without aborting (e.g. run inline and return) since `last_msg` is an
internal scheduling hint.
--
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]
