diff --git a/src/brpc/input_messenger.cpp b/src/brpc/input_messenger.cpp index 81154be134..f47b30646f 100644 --- a/src/brpc/input_messenger.cpp +++ b/src/brpc/input_messenger.cpp @@ -17,6 +17,8 @@ #include +#include +#include #include "butil/fd_guard.h" // fd_guard #include "butil/logging.h" // CHECK #include "butil/time.h" // cpuwide_time_us @@ -72,6 +74,17 @@ DEFINE_int32(socket_tcp_user_timeout_ms, -1, "connection and return ETIMEDOUT to the application. Only linux supports " "TCP_USER_TIMEOUT."); +DEFINE_int32(input_message_batch_process_size, 0, + "Experimental. -1 adaptively processes up to 16 parsed input " + "messages in one bthread based on the recent per-socket burst. " + "Values greater than 1 use a fixed batch size. 0 or 1 preserves " + "the original one-message-per-bthread behavior."); +static bool ValidateInputMessageBatchProcessSize(const char*, int32_t value) { + return value >= -1; +} +BRPC_VALIDATE_GFLAG(input_message_batch_process_size, + ValidateInputMessageBatchProcessSize); + DECLARE_bool(usercode_in_pthread); DECLARE_bool(usercode_in_coroutine); DECLARE_uint64(max_body_size); @@ -79,6 +92,10 @@ DECLARE_uint64(max_body_size); const size_t MSG_SIZE_WINDOW = 10; // Take last so many message into stat. const size_t MIN_ONCE_READ = 4096; const size_t MAX_ONCE_READ = 524288; +const uint32_t INPUT_BATCH_EMA_SCALE = 256; +const uint32_t MAX_ADAPTIVE_INPUT_BATCH_SIZE = 16; +const uint32_t MAX_ADAPTIVE_INPUT_BATCH_SAMPLE = + MAX_ADAPTIVE_INPUT_BATCH_SIZE * 2; ParseResult InputMessenger::CutInputMessage( Socket* m, size_t* index, bool read_eof) { @@ -173,6 +190,30 @@ void* ProcessInputMessage(void* void_arg) { return nullptr; } +void* ProcessInputMessageBatch(void* void_arg) { + std::unique_ptr batch( + static_cast(void_arg)); + batch->Run(); + return nullptr; +} + +InputMessageBatch::~InputMessageBatch() noexcept(false) { + Run(); +} + +void InputMessageBatch::add(InputMessageBase* msg) { + if (msg) { + _msgs.push_back(msg); + } +} + +void InputMessageBatch::Run() { + for (size_t i = 0; i < _msgs.size(); ++i) { + ProcessInputMessage(_msgs[i]); + } + _msgs.clear(); +} + struct RunLastMessage { inline void operator()(InputMessageBase* last_msg) { ProcessInputMessage(last_msg); @@ -192,6 +233,79 @@ void InputMessageClosure::reset(InputMessageBase* m) { _msg = m; } +void InputMessenger::QueueInputMessageBatch( + Socket* m, std::unique_ptr* 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* 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)); + } + if (!batch->get()) { + last_msg.reset(msg); + m->_transport->QueueMessage( + last_msg, num_bthread_created, false); + return; + } + (*batch)->add(msg); + if ((*batch)->size() >= batch_size) { + QueueInputMessageBatch( + m, batch, num_bthread_created, false); + } +} + +uint32_t InputMessenger::UpdateAdaptiveBatchSize( + uint32_t* messages_per_read_ema_q8, + uint32_t current_batch_size, + size_t parsed_message_count) { + if (parsed_message_count == 0) { + return current_batch_size; + } + if (*messages_per_read_ema_q8 == 0 || current_batch_size == 0) { + *messages_per_read_ema_q8 = INPUT_BATCH_EMA_SCALE; + current_batch_size = 1; + } + + const uint32_t sample = static_cast( + std::min(parsed_message_count, + static_cast(MAX_ADAPTIVE_INPUT_BATCH_SAMPLE))); + const uint32_t sample_q8 = sample * INPUT_BATCH_EMA_SCALE; + uint32_t ema_q8 = *messages_per_read_ema_q8; + if (sample_q8 > ema_q8) { + // Increase slowly to avoid turning a short burst into persistent + // head-of-line blocking. + ema_q8 += (sample_q8 - ema_q8) / 8; + } else { + // Reduce quickly when the connection becomes sparse. + ema_q8 -= (ema_q8 - sample_q8 + 1) / 2; + } + *messages_per_read_ema_q8 = ema_q8; + + uint32_t desired_batch_size = 1; + while (desired_batch_size < MAX_ADAPTIVE_INPUT_BATCH_SIZE && + ema_q8 > desired_batch_size * INPUT_BATCH_EMA_SCALE) { + desired_batch_size *= 2; + } + if (desired_batch_size > current_batch_size) { + // Increase at most one level for each observation. + return std::min(current_batch_size * 2, desired_batch_size); + } + return desired_batch_size; +} + int InputMessenger::ProcessNewMessage( Socket* m, ssize_t bytes, bool read_eof, const uint64_t received_us, const uint64_t base_realtime, @@ -203,6 +317,27 @@ int InputMessenger::ProcessNewMessage( size_t last_size = m->_read_buf.length(); int num_bthread_created = 0; + const int configured_batch_size = + FLAGS_input_message_batch_process_size; + const bool adaptive_batch_process = + configured_batch_size == -1 && !FLAGS_usercode_in_coroutine; + size_t batch_size = configured_batch_size > 0 + ? static_cast(configured_batch_size) : 1; + if (adaptive_batch_process) { + if (m->_adaptive_input_message_batch_size == 0) { + m->_input_messages_per_read_ema_q8 = INPUT_BATCH_EMA_SCALE; + m->_adaptive_input_message_batch_size = 1; + } + batch_size = m->_adaptive_input_message_batch_size; + } else if (m->_adaptive_input_message_batch_size != 0) { + // Do not reuse history after switching away from adaptive mode. + m->_input_messages_per_read_ema_q8 = 0; + m->_adaptive_input_message_batch_size = 0; + } + const bool batch_process = + batch_size > 1 && !FLAGS_usercode_in_coroutine; + size_t batchable_message_count = 0; + std::unique_ptr input_batch; while (1) { size_t index = 8888; ParseResult pr = CutInputMessage(m, &index, read_eof); @@ -258,7 +393,14 @@ int InputMessenger::ProcessNewMessage( // This unique_ptr prevents msg to be lost before transfering // ownership to last_msg DestroyingPtr msg(pr.message()); - m->_transport->QueueMessage(last_msg, &num_bthread_created, false); + if (batch_process) { + QueueLastMessageOrBatch( + m, last_msg, &input_batch, + &num_bthread_created, batch_size); + } else { + m->_transport->QueueMessage( + last_msg, &num_bthread_created, false); + } if (_handlers[index].process == nullptr) { LOG(ERROR) << "process of index=" << index << " is NULL"; continue; @@ -290,8 +432,15 @@ int InputMessenger::ProcessNewMessage( if (!m->is_read_progressive()) { // Transfer ownership to last_msg last_msg.reset(msg.release()); + if (adaptive_batch_process) { + ++batchable_message_count; + } } else { last_msg.reset(msg.release()); + if (batch_process) { + QueueInputMessageBatch( + m, &input_batch, &num_bthread_created, false); + } m->_transport->QueueMessage(last_msg, &num_bthread_created, false); bthread_flush(); num_bthread_created = 0; @@ -301,9 +450,21 @@ 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, false); + } + if (m->_socket_mode == SOCKET_MODE_RDMA || + m->_socket_mode == SOCKET_MODE_UBRING) { m->_transport->QueueMessage(last_msg, &num_bthread_created, true); } + if (adaptive_batch_process && batchable_message_count != 0) { + m->_adaptive_input_message_batch_size = + UpdateAdaptiveBatchSize( + &m->_input_messages_per_read_ema_q8, + m->_adaptive_input_message_batch_size, + batchable_message_count); + } if (num_bthread_created) { bthread_flush(); } @@ -317,8 +478,10 @@ void InputMessenger::OnNewMessages(Socket* m) { // - 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. To minimize the overhead, scheduling - // is batched(notice the BTHREAD_NOSIGNAL and bthread_flush). + // processed in separate bthreads, or in batches when + // -input_message_batch_process_size is -1 or greater than 1. To minimize + // the overhead, scheduling is batched(notice the BTHREAD_NOSIGNAL and + // bthread_flush). // - Verify will always be called in this bthread at most once and before // any process. InputMessenger* messenger = static_cast(m->user()); diff --git a/src/brpc/input_messenger.h b/src/brpc/input_messenger.h index d056263e2e..4fb1560945 100644 --- a/src/brpc/input_messenger.h +++ b/src/brpc/input_messenger.h @@ -19,6 +19,9 @@ #ifndef BRPC_INPUT_MESSENGER_H #define BRPC_INPUT_MESSENGER_H +#include +#include + #include "butil/iobuf.h" // butil::IOBuf #include "brpc/socket.h" // SocketId, SocketUser #include "brpc/parse_result.h" // ParseResult @@ -91,6 +94,26 @@ class InputMessageClosure { InputMessageBase* _msg; }; +class InputMessageBatch { +public: + InputMessageBatch() {} + explicit InputMessageBatch(size_t capacity) { + _msgs.reserve(capacity); + } + ~InputMessageBatch() noexcept(false); + + void add(InputMessageBase* msg); + void Run(); + bool empty() const { return _msgs.empty(); } + size_t size() const { return _msgs.size(); } + +private: + std::vector _msgs; +}; + +void* ProcessInputMessage(void* void_arg); +void* ProcessInputMessageBatch(void* void_arg); + // Process messages from connections. // `Message' corresponds to a client's request or a server's response. class InputMessenger : public SocketUser { @@ -136,7 +159,6 @@ friend class ubring::UBShmEndpoint; static void OnNewMessages(Socket* m); private: - // Find a valid scissor from `handlers' to cut off `header' and `payload' // from m->read_buf, save index of the scissor into `index'. ParseResult CutInputMessage(Socket* m, size_t* index, bool read_eof); @@ -148,6 +170,20 @@ friend class ubring::UBShmEndpoint; const uint64_t received_us, const uint64_t base_realtime, InputMessageClosure& last_msg); + static void QueueInputMessageBatch( + Socket* m, std::unique_ptr* batch, + int* num_bthread_created, bool last_msg); + + static void QueueLastMessageOrBatch( + Socket* m, InputMessageClosure& last_msg, + std::unique_ptr* batch, + int* num_bthread_created, size_t batch_size); + + static uint32_t UpdateAdaptiveBatchSize( + uint32_t* messages_per_read_ema_q8, + uint32_t current_batch_size, + size_t parsed_message_count); + // User-supplied scissors and handlers. // the index of handler is exactly the same as the protocol InputMessageHandler* _handlers; diff --git a/src/brpc/rdma_transport.cpp b/src/brpc/rdma_transport.cpp index ee5151c3a5..674d57f1c3 100644 --- a/src/brpc/rdma_transport.cpp +++ b/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; + } + + 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; + } +} + void RdmaTransport::Debug(std::ostream &os) { if (_rdma_state == RDMA_ON && _rdma_ep) { _rdma_ep->DebugInfo(os); diff --git a/src/brpc/rdma_transport.h b/src/brpc/rdma_transport.h index 1d78fbb430..bf60bcef6c 100644 --- a/src/brpc/rdma_transport.h +++ b/src/brpc/rdma_transport.h @@ -40,6 +40,9 @@ friend class rdma::RdmaHandshakeServerV3; int WaitEpollOut(butil::atomic* _epollout_butex, bool pollin, const timespec duetime) override; void ProcessEvent(bthread_attr_t attr) override; void QueueMessage(InputMessageClosure& inputMsg, int* num_bthread_created, bool last_msg) override; + void QueueMessages(InputMessageBatch* inputMsgs, + int* num_bthread_created, + bool last_msg) override; void Debug(std::ostream &os) override; rdma::RdmaEndpoint* GetRdmaEp() { CHECK(_rdma_ep != nullptr); @@ -64,4 +67,4 @@ friend class rdma::RdmaHandshakeServerV3; }; } // namespace brpc #endif // BRPC_WITH_RDMA -#endif //BRPC_RDMA_TRANSPORT_H \ No newline at end of file +#endif //BRPC_RDMA_TRANSPORT_H diff --git a/src/brpc/socket.cpp b/src/brpc/socket.cpp index 38727571ba..e7118c49e8 100644 --- a/src/brpc/socket.cpp +++ b/src/brpc/socket.cpp @@ -467,6 +467,8 @@ Socket::Socket(Forbidden f) , _hc_count(0) , _last_msg_size(0) , _avg_msg_size(0) + , _input_messages_per_read_ema_q8(0) + , _adaptive_input_message_batch_size(0) , _last_readtime_us(0) , _parsing_context(nullptr) , _correlation_id(0) @@ -571,9 +573,11 @@ void Socket::ReleaseAllFailedWriteRequests(Socket::WriteRequest* req) { } int Socket::ResetFileDescriptor(int fd) { - // Reset message sizes when fd is changed. + // Reset input heuristics when fd is changed. _last_msg_size = 0; _avg_msg_size = 0; + _input_messages_per_read_ema_q8 = 0; + _adaptive_input_message_batch_size = 0; // MUST store `_fd' before adding itself into epoll device to avoid // race conditions with the callback function inside epoll static butil::atomic BAIDU_CACHELINE_ALIGNMENT fd_version(0); @@ -2377,6 +2381,10 @@ void Socket::DebugSocket(std::ostream& os, SocketId id) { const int64_t cpuwide_now = butil::cpuwide_time_us(); os << "\nhc_count=" << ptr->_hc_count << "\navg_input_msg_size=" << ptr->_avg_msg_size + << "\navg_input_messages_per_read=" + << ((ptr->_input_messages_per_read_ema_q8 + 128) >> 8) + << "\nadaptive_input_message_batch_size=" + << ptr->_adaptive_input_message_batch_size // NOTE: We're assuming that butil::IOBuf.size() is thread-safe, it is now // however it's not guaranteed. << "\nread_buf=" << ptr->_read_buf.size() diff --git a/src/brpc/socket.h b/src/brpc/socket.h index 6d321f8bc3..e9d25775d5 100644 --- a/src/brpc/socket.h +++ b/src/brpc/socket.h @@ -898,6 +898,11 @@ friend class TransportFactory; uint32_t _last_msg_size; // Average message size of last #MSG_SIZE_WINDOW messages (roughly) uint32_t _avg_msg_size; + // Q8 EMA of processable messages parsed in one read. Accessed only from + // the serialized input callback. + uint32_t _input_messages_per_read_ema_q8; + // 0 when adaptive batching is inactive, otherwise one of 1/2/4/8/16. + uint32_t _adaptive_input_message_batch_size; // Storing data read from `_fd' but cut-off yet. butil::IOPortal _read_buf; diff --git a/src/brpc/tcp_transport.cpp b/src/brpc/tcp_transport.cpp index 98fea81674..b6016a0e2d 100644 --- a/src/brpc/tcp_transport.cpp +++ b/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, bool) { + 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; + } +} + } // namespace brpc diff --git a/src/brpc/tcp_transport.h b/src/brpc/tcp_transport.h index 8a06a85d37..2215281da4 100644 --- a/src/brpc/tcp_transport.h +++ b/src/brpc/tcp_transport.h @@ -34,8 +34,11 @@ class TcpTransport : public Transport { int WaitEpollOut(butil::atomic* _epollout_butex, bool pollin, timespec duetime) override; void ProcessEvent(bthread_attr_t attr) override; void QueueMessage(InputMessageClosure& input_msg, int* num_bthread_created, bool last_msg) override; + void QueueMessages(InputMessageBatch* input_msgs, + int* num_bthread_created, + bool last_msg) override; void Debug(std::ostream &os) override {} }; } // namespace brpc -#endif //BRPC_TCP_TRANSPORT_H \ No newline at end of file +#endif //BRPC_TCP_TRANSPORT_H diff --git a/src/brpc/transport.h b/src/brpc/transport.h index edef24879c..ac5d49ac70 100644 --- a/src/brpc/transport.h +++ b/src/brpc/transport.h @@ -49,6 +49,9 @@ class Transport { virtual int WaitEpollOut(butil::atomic* _epollout_butex, bool pollin, timespec duetime) = 0; virtual void ProcessEvent(bthread_attr_t attr) = 0; virtual void QueueMessage(InputMessageClosure& input_msg, int* num_bthread_created, bool last_msg) = 0; + virtual void QueueMessages(InputMessageBatch* input_msgs, + int* num_bthread_created, + bool last_msg) = 0; virtual void Debug(std::ostream &os) = 0; bool HasOnEdgeTrigger() { @@ -63,4 +66,4 @@ class Transport { OnEdgeTrigger _on_edge_trigger; }; } -#endif //BRPC_TRANSPORT_H \ No newline at end of file +#endif //BRPC_TRANSPORT_H diff --git a/src/brpc/ubshm_transport.cpp b/src/brpc/ubshm_transport.cpp index df4eb36bed..3352bfd837 100644 --- a/src/brpc/ubshm_transport.cpp +++ b/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; + } + + 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(); + bthread_attr_set_name(&tmp, "ProcessInputMessageBatch"); + if (!FLAGS_usercode_in_coroutine && bthread_start_background( + &th, &tmp, ProcessInputMessageBatch, input_msgs) == 0) { + ++*num_bthread_created; + } else { + input_msgs->Run(); + delete input_msgs; + } +} + void UBShmTransport::Debug(std::ostream &os) {} int UBShmTransport::ContextInitOrDie(bool serverOrNot, const void* _options) { @@ -232,4 +261,4 @@ bool UBShmTransport::OptionsAvailableOverUB(const ServerOptions* opt) { return true; } } // namespace brpc -#endif \ No newline at end of file +#endif diff --git a/src/brpc/ubshm_transport.h b/src/brpc/ubshm_transport.h index b3d1e7c518..7a29ed2f5c 100644 --- a/src/brpc/ubshm_transport.h +++ b/src/brpc/ubshm_transport.h @@ -37,6 +37,9 @@ friend class ubring::UBConnect; int WaitEpollOut(butil::atomic* _epollout_butex, bool pollin, const timespec duetime) override; void ProcessEvent(bthread_attr_t attr) override; void QueueMessage(InputMessageClosure& inputMsg, int* num_bthread_created, bool last_msg) override; + void QueueMessages(InputMessageBatch* input_msgs, + int* num_bthread_created, + bool last_msg) override; void Debug(std::ostream &os) override; ubring::UBShmEndpoint* GetUBShmEp() { CHECK(_ub_ep != nullptr); @@ -61,4 +64,4 @@ friend class ubring::UBConnect; }; } // namespace brpc #endif // BRPC_WITH_UBRING -#endif //BRPC_UB_TRANSPORT_H \ No newline at end of file +#endif //BRPC_UB_TRANSPORT_H