You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by al...@apache.org on 2026/05/26 17:19:54 UTC
(kudu) branch master updated: KUDU-1865 reduce cross-thread allocations in RPC (part 2)
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
The following commit(s) were added to refs/heads/master by this push:
new 1a048343e KUDU-1865 reduce cross-thread allocations in RPC (part 2)
1a048343e is described below
commit 1a048343ec7587d70c29313e25a4ccba872f7108
Author: Alexey Serbin <al...@apache.org>
AuthorDate: Tue May 19 21:59:14 2026 -0700
KUDU-1865 reduce cross-thread allocations in RPC (part 2)
With this update, the OutboundCall::call_response_ member field
of the CallResponse type is no longer allocated on the heap. Instead,
it's allocated on the stack and the move semantics is used to pass
the object around.
I ran the same test scenario as in the 'part 1' to trace the asymmetry
in tcmalloc's cross-thread allocations/deallocations and confirmed
that the entries related to CallResponse were gone.
I also ran the test scenario to account for the total number of updates
in tcmalloc's central free lists. This patch results in ~1.07x times
reduction in the total number of updates in the central free lists with
the default setting of TCMALLOC_MAX_TOTAL_THREAD_CACHE_BYTES:
before: 367
after: 341
The requests-per-second performance hasn't changed per reports from
the RpcBench.BenchmarkCalls scenario.
Change-Id: Ia5b8771f4092d5e566e91b7c7fdbf560c5fd3851
Reviewed-on: http://gerrit.cloudera.org:8080/24329
Tested-by: Alexey Serbin <al...@apache.org>
Reviewed-by: Michael Smith <mi...@cloudera.com>
Reviewed-by: Abhishek Chennaka <ac...@cloudera.com>
---
src/kudu/rpc/connection.cc | 10 ++--
src/kudu/rpc/outbound_call.cc | 12 ++--
src/kudu/rpc/outbound_call.h | 129 +++++++++++++++++++++--------------------
src/kudu/rpc/rpc_controller.cc | 4 +-
4 files changed, 77 insertions(+), 78 deletions(-)
diff --git a/src/kudu/rpc/connection.cc b/src/kudu/rpc/connection.cc
index fa2a9743b..3d87cdc81 100644
--- a/src/kudu/rpc/connection.cc
+++ b/src/kudu/rpc/connection.cc
@@ -637,15 +637,15 @@ void Connection::HandleIncomingCall(unique_ptr<InboundTransfer> transfer) {
void Connection::HandleCallResponse(unique_ptr<InboundTransfer> transfer) {
DCHECK(reactor_thread_->IsCurrentThread());
- unique_ptr<CallResponse> resp(new CallResponse);
- CHECK_OK(resp->ParseFrom(std::move(transfer)));
+ CallResponse resp;
+ CHECK_OK(resp.ParseFrom(std::move(transfer)));
CallAwaitingResponse* car_ptr = EraseKeyReturnValuePtr(
- &awaiting_response_, resp->call_id());
+ &awaiting_response_, resp.call_id());
if (PREDICT_FALSE(car_ptr == nullptr)) {
LOG(WARNING) << Substitute(
"$0: got a response for call id $1 which was not pending, ignoring",
- ToString(), resp->call_id());
+ ToString(), resp.call_id());
return;
}
@@ -656,7 +656,7 @@ void Connection::HandleCallResponse(unique_ptr<InboundTransfer> transfer) {
// The call already failed due to a timeout.
VLOG(1) << Substitute(
"got response to call id $0 after client already timed out or cancelled",
- resp->call_id());
+ resp.call_id());
return;
}
diff --git a/src/kudu/rpc/outbound_call.cc b/src/kudu/rpc/outbound_call.cc
index 228a1e091..eb94ea002 100644
--- a/src/kudu/rpc/outbound_call.cc
+++ b/src/kudu/rpc/outbound_call.cc
@@ -323,11 +323,11 @@ void OutboundCall::CallCallback() {
}
}
-void OutboundCall::SetResponse(unique_ptr<CallResponse> resp) {
+void OutboundCall::SetResponse(CallResponse&& resp) {
call_response_ = std::move(resp);
- Slice r(call_response_->serialized_response());
+ Slice r(call_response_.serialized_response());
- if (call_response_->is_success()) {
+ if (call_response_.is_success()) {
// TODO: here we're deserializing the call response within the reactor thread,
// which isn't great, since it would block processing of other RPCs in parallel.
// Should look into a way to avoid this.
@@ -532,10 +532,6 @@ void OutboundCall::DumpPB(const DumpConnectionsRequestPB& req,
/// CallResponse
///
-CallResponse::CallResponse()
- : parsed_(false) {
-}
-
Status CallResponse::GetSidecar(int idx, Slice* sidecar) const {
DCHECK(parsed_);
if (PREDICT_FALSE(idx < 0 || idx >= header_.sidecar_offsets_size())) {
@@ -547,7 +543,7 @@ Status CallResponse::GetSidecar(int idx, Slice* sidecar) const {
}
Status CallResponse::ParseFrom(unique_ptr<InboundTransfer> transfer) {
- CHECK(!parsed_);
+ DCHECK(!parsed_);
RETURN_NOT_OK(serialization::ParseMessage(transfer->data(), &header_,
&serialized_response_));
diff --git a/src/kudu/rpc/outbound_call.h b/src/kudu/rpc/outbound_call.h
index 620a28b43..683ff550c 100644
--- a/src/kudu/rpc/outbound_call.h
+++ b/src/kudu/rpc/outbound_call.h
@@ -23,6 +23,7 @@
#include <string>
#include <vector>
+#include <boost/container/small_vector.hpp>
#include <gflags/gflags_declare.h>
#include <glog/logging.h>
#include <gtest/gtest_prod.h>
@@ -52,7 +53,6 @@ class Message;
namespace kudu {
namespace rpc {
-class CallResponse;
class DumpConnectionsRequestPB;
class RpcCallInProgressPB;
class RpcController;
@@ -94,6 +94,69 @@ class RequestPayload {
int32_t sidecar_byte_size_ = -1;
};
+// A response to a call, on the client side.
+// Upon receiving a response, this is allocated in the reactor thread and filled
+// into the OutboundCall instance via OutboundCall::SetResponse.
+//
+// This may either be a success or error response.
+//
+// This class takes care of separating out the distinct payload slices sent
+// over.
+class CallResponse final {
+ public:
+ CallResponse()
+ : parsed_(false) {
+ }
+
+ CallResponse& operator=(CallResponse&& other) noexcept = default;
+
+ // Parse the response received from a call. This must be called before any
+ // other methods on this object.
+ Status ParseFrom(std::unique_ptr<InboundTransfer> transfer);
+
+ // Return true if the call succeeded.
+ bool is_success() const {
+ DCHECK(parsed_);
+ return !header_.is_error();
+ }
+
+ // Return the call ID that this response is related to.
+ int32_t call_id() const {
+ DCHECK(parsed_);
+ return header_.call_id();
+ }
+
+ // Return the serialized response data. This is just the response "body" --
+ // either a serialized ErrorStatusPB, or the serialized user response protobuf.
+ const Slice &serialized_response() const {
+ DCHECK(parsed_);
+ return serialized_response_;
+ }
+
+ // See RpcController::GetSidecar()
+ Status GetSidecar(int idx, Slice* sidecar) const;
+
+ private:
+ // True once ParseFrom() is called.
+ bool parsed_;
+
+ // The parsed header.
+ ResponseHeader header_;
+
+ // The slice of data for the encoded protobuf response.
+ // This slice refers to memory allocated by transfer_
+ Slice serialized_response_;
+
+ // Slices of data for rpc sidecars. They point into memory owned by transfer_.
+ SidecarSliceVector sidecar_slices_;
+
+ // The incoming transfer data - retained because serialized_response_
+ // and sidecar_slices_ refer into its data.
+ std::unique_ptr<InboundTransfer> transfer_;
+
+ DISALLOW_COPY_AND_ASSIGN(CallResponse);
+};
+
// Tracks the status of a call on the client side.
//
// This is an internal-facing class -- clients interact with the
@@ -193,7 +256,7 @@ class OutboundCall {
bool IsFinished() const;
// Fill in the call response.
- void SetResponse(std::unique_ptr<CallResponse> resp);
+ void SetResponse(CallResponse&& resp);
const std::set<RpcFeatureFlag>& required_rpc_features() const {
return required_rpc_features_;
@@ -319,8 +382,7 @@ class OutboundCall {
std::unique_ptr<RequestPayload> payload_;
// Once a response has been received for this call, contains that response.
- // Otherwise NULL.
- std::unique_ptr<CallResponse> call_response_;
+ CallResponse call_response_;
// True if cancellation was requested on this call.
bool cancellation_requested_;
@@ -328,64 +390,5 @@ class OutboundCall {
DISALLOW_COPY_AND_ASSIGN(OutboundCall);
};
-// A response to a call, on the client side.
-// Upon receiving a response, this is allocated in the reactor thread and filled
-// into the OutboundCall instance via OutboundCall::SetResponse.
-//
-// This may either be a success or error response.
-//
-// This class takes care of separating out the distinct payload slices sent
-// over.
-class CallResponse {
- public:
- CallResponse();
-
- // Parse the response received from a call. This must be called before any
- // other methods on this object.
- Status ParseFrom(std::unique_ptr<InboundTransfer> transfer);
-
- // Return true if the call succeeded.
- bool is_success() const {
- DCHECK(parsed_);
- return !header_.is_error();
- }
-
- // Return the call ID that this response is related to.
- int32_t call_id() const {
- DCHECK(parsed_);
- return header_.call_id();
- }
-
- // Return the serialized response data. This is just the response "body" --
- // either a serialized ErrorStatusPB, or the serialized user response protobuf.
- const Slice &serialized_response() const {
- DCHECK(parsed_);
- return serialized_response_;
- }
-
- // See RpcController::GetSidecar()
- Status GetSidecar(int idx, Slice* sidecar) const;
-
- private:
- // True once ParseFrom() is called.
- bool parsed_;
-
- // The parsed header.
- ResponseHeader header_;
-
- // The slice of data for the encoded protobuf response.
- // This slice refers to memory allocated by transfer_
- Slice serialized_response_;
-
- // Slices of data for rpc sidecars. They point into memory owned by transfer_.
- SidecarSliceVector sidecar_slices_;
-
- // The incoming transfer data - retained because serialized_response_
- // and sidecar_slices_ refer into its data.
- std::unique_ptr<InboundTransfer> transfer_;
-
- DISALLOW_COPY_AND_ASSIGN(CallResponse);
-};
-
} // namespace rpc
} // namespace kudu
diff --git a/src/kudu/rpc/rpc_controller.cc b/src/kudu/rpc/rpc_controller.cc
index b362f40b7..effc1cc5e 100644
--- a/src/kudu/rpc/rpc_controller.cc
+++ b/src/kudu/rpc/rpc_controller.cc
@@ -105,11 +105,11 @@ const ErrorStatusPB* RpcController::error_response() const {
}
int32_t RpcController::call_id() const {
- return call_->call_response_->call_id();
+ return call_->call_response_.call_id();
}
Status RpcController::GetInboundSidecar(int idx, Slice* sidecar) const {
- return call_->call_response_->GetSidecar(idx, sidecar);
+ return call_->call_response_.GetSidecar(idx, sidecar);
}
void RpcController::set_timeout(const MonoDelta& timeout) {