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/04/08 02:01:00 UTC
(kudu) branch master updated: [rpc] add PREDICT_FALSE() for unlikely error scenarios
This is an automated email from the ASF dual-hosted git repository.
alexey 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 748ef1053 [rpc] add PREDICT_FALSE() for unlikely error scenarios
748ef1053 is described below
commit 748ef105309d0a5423660ea100f33ec56fb5cc45
Author: Alexey Serbin <al...@apache.org>
AuthorDate: Mon Apr 6 21:20:15 2026 -0700
[rpc] add PREDICT_FALSE() for unlikely error scenarios
I was looking at the RPC code to clarify a few perf-related questions
and found that PREDICT_{TRUE,FALSE}() were not used uniformly throughout
the code. This patch adds these macros in the rest of relevant places,
where applicable.
This changelist doesn't contain any functional modifications.
Change-Id: I3178a37ec371ca4db2b818a67cf04f52b72f2dd0
Reviewed-on: http://gerrit.cloudera.org:8080/24173
Reviewed-by: Abhishek Chennaka <ac...@cloudera.com>
Tested-by: Alexey Serbin <al...@apache.org>
---
src/kudu/rpc/client_negotiation.cc | 6 +++---
src/kudu/rpc/client_negotiation.h | 7 +++----
src/kudu/rpc/connection.cc | 6 +++---
src/kudu/rpc/inbound_call.cc | 15 ++++++++-------
src/kudu/rpc/messenger.cc | 2 +-
src/kudu/rpc/negotiation.cc | 19 +++++++++----------
src/kudu/rpc/outbound_call.cc | 4 ++--
src/kudu/rpc/proxy.cc | 4 ++--
src/kudu/rpc/reactor.cc | 8 ++++----
src/kudu/rpc/rpc_controller.cc | 7 ++++---
src/kudu/rpc/rpc_sidecar.cc | 11 ++++++-----
src/kudu/rpc/serialization.cc | 2 +-
src/kudu/rpc/service_pool.cc | 2 +-
13 files changed, 47 insertions(+), 46 deletions(-)
diff --git a/src/kudu/rpc/client_negotiation.cc b/src/kudu/rpc/client_negotiation.cc
index 42c23f099..f65d2035d 100644
--- a/src/kudu/rpc/client_negotiation.cc
+++ b/src/kudu/rpc/client_negotiation.cc
@@ -275,7 +275,7 @@ Status ClientNegotiation::RecvNegotiatePB(NegotiatePB* msg,
Status ClientNegotiation::ParseError(const Slice& err_data,
unique_ptr<ErrorStatusPB>* rpc_error) {
unique_ptr<ErrorStatusPB> error(new ErrorStatusPB);
- if (!error->ParseFromArray(err_data.data(), err_data.size())) {
+ if (PREDICT_FALSE(!error->ParseFromArray(err_data.data(), err_data.size()))) {
return Status::IOError("invalid error response, missing fields",
error->InitializationErrorString());
}
@@ -403,8 +403,8 @@ Status ClientNegotiation::HandleNegotiate(const NegotiatePB& response) {
}
}
- if (encryption_ == RpcEncryption::REQUIRED &&
- !ContainsKey(server_features_, RpcFeatureFlag::TLS)) {
+ if (PREDICT_FALSE(encryption_ == RpcEncryption::REQUIRED &&
+ !ContainsKey(server_features_, RpcFeatureFlag::TLS))) {
return Status::NotAuthorized("server does not support required TLS encryption");
}
diff --git a/src/kudu/rpc/client_negotiation.h b/src/kudu/rpc/client_negotiation.h
index 2fd2ffb68..44f51aef4 100644
--- a/src/kudu/rpc/client_negotiation.h
+++ b/src/kudu/rpc/client_negotiation.h
@@ -146,6 +146,9 @@ class ClientNegotiation {
static Status CheckGSSAPI() WARN_UNUSED_RESULT;
private:
+ // Parse error status message from raw bytes of an ErrorStatusPB.
+ static Status ParseError(const Slice& err_data,
+ std::unique_ptr<ErrorStatusPB>* rpc_error) WARN_UNUSED_RESULT;
// Encode and send the specified negotiate request message to the server.
Status SendNegotiatePB(const NegotiatePB& msg) WARN_UNUSED_RESULT;
@@ -156,10 +159,6 @@ class ClientNegotiation {
faststring* buffer,
std::unique_ptr<ErrorStatusPB>* rpc_error) WARN_UNUSED_RESULT;
- // Parse error status message from raw bytes of an ErrorStatusPB.
- Status ParseError(const Slice& err_data,
- std::unique_ptr<ErrorStatusPB>* rpc_error) WARN_UNUSED_RESULT;
-
Status SendConnectionHeader() WARN_UNUSED_RESULT;
// Initialize the SASL client negotiation instance.
diff --git a/src/kudu/rpc/connection.cc b/src/kudu/rpc/connection.cc
index 02ed68676..e157f7db0 100644
--- a/src/kudu/rpc/connection.cc
+++ b/src/kudu/rpc/connection.cc
@@ -515,7 +515,7 @@ void Connection::ReadHandler(ev::io& /*watcher*/, int revents) {
DCHECK(reactor_thread_->IsCurrentThread());
DVLOG(3) << Substitute("$0 ReadHandler(revents=$1)", ToString(), revents);
- if (revents & EV_ERROR) {
+ if (PREDICT_FALSE(revents & EV_ERROR)) {
reactor_thread_->DestroyConnection(this, Status::NetworkError(ToString() +
": ReadHandler encountered an error"));
return;
@@ -633,14 +633,14 @@ void Connection::HandleCallResponse(unique_ptr<InboundTransfer> transfer) {
void Connection::WriteHandler(ev::io& /*watcher*/, int revents) {
DCHECK(reactor_thread_->IsCurrentThread());
- if (revents & EV_ERROR) {
+ if (PREDICT_FALSE(revents & EV_ERROR)) {
reactor_thread_->DestroyConnection(this, Status::NetworkError(ToString() +
": writeHandler encountered an error"));
return;
}
DVLOG(3) << Substitute("$0: writeHandler: revents=$1", ToString(), revents);
- if (outbound_transfers_.empty()) {
+ if (PREDICT_FALSE(outbound_transfers_.empty())) {
LOG(WARNING) << Substitute(
"$0 got a ready-to-write callback, but there is nothing to write",
ToString());
diff --git a/src/kudu/rpc/inbound_call.cc b/src/kudu/rpc/inbound_call.cc
index 18ac13582..8fa760d96 100644
--- a/src/kudu/rpc/inbound_call.cc
+++ b/src/kudu/rpc/inbound_call.cc
@@ -97,8 +97,9 @@ Status InboundCall::ParseFrom(unique_ptr<InboundTransfer> transfer) {
deadline_ = timing_.time_received + MonoDelta::FromMilliseconds(header_.timeout_millis());
}
- if (header_.sidecar_offsets_size() > TransferLimits::kMaxSidecars) {
- return Status::Corruption(strings::Substitute(
+ if (PREDICT_FALSE(header_.sidecar_offsets_size() >
+ TransferLimits::kMaxSidecars)) {
+ return Status::Corruption(Substitute(
"Received $0 additional payload slices, expected at most %d",
header_.sidecar_offsets_size(), TransferLimits::kMaxSidecars));
}
@@ -224,12 +225,12 @@ Status InboundCall::AddOutboundSidecar(unique_ptr<RpcSidecar> car, int* idx) {
// Check that the number of sidecars does not exceed the number of payload
// slices that are free (two are used up by the header and main message
// protobufs).
- if (outbound_sidecars_.size() > TransferLimits::kMaxSidecars) {
+ if (PREDICT_FALSE(outbound_sidecars_.size() > TransferLimits::kMaxSidecars)) {
return Status::ServiceUnavailable("All available sidecars already used");
}
size_t sidecar_bytes = car->TotalSize();
- if (outbound_sidecars_total_bytes_ >
- TransferLimits::kMaxTotalSidecarBytes - sidecar_bytes) {
+ if (PREDICT_FALSE(outbound_sidecars_total_bytes_ + sidecar_bytes >
+ TransferLimits::kMaxTotalSidecarBytes)) {
return Status::RuntimeError(Substitute("Total size of sidecars $0 would exceed limit $1",
static_cast<int64_t>(outbound_sidecars_total_bytes_) + sidecar_bytes,
TransferLimits::kMaxTotalSidecarBytes));
@@ -349,8 +350,8 @@ vector<uint32_t> InboundCall::GetRequiredFeatures() const {
Status InboundCall::GetInboundSidecar(int idx, Slice* sidecar) const {
DCHECK(transfer_) << "Sidecars have been discarded";
- if (idx < 0 || idx >= header_.sidecar_offsets_size()) {
- return Status::InvalidArgument(strings::Substitute(
+ if (PREDICT_FALSE(idx < 0 || idx >= header_.sidecar_offsets_size())) {
+ return Status::InvalidArgument(Substitute(
"Index $0 does not reference a valid sidecar", idx));
}
*sidecar = inbound_sidecar_slices_[idx];
diff --git a/src/kudu/rpc/messenger.cc b/src/kudu/rpc/messenger.cc
index a4020da58..20da3d1e5 100644
--- a/src/kudu/rpc/messenger.cc
+++ b/src/kudu/rpc/messenger.cc
@@ -485,7 +485,7 @@ const scoped_refptr<RpcService> Messenger::rpc_service(const string& service_nam
scoped_refptr<RpcService> service;
{
shared_lock guard(lock_.get_lock());
- if (!FindCopy(rpc_services_, service_name, &service)) {
+ if (PREDICT_FALSE(!FindCopy(rpc_services_, service_name, &service))) {
return scoped_refptr<RpcService>(nullptr);
}
}
diff --git a/src/kudu/rpc/negotiation.cc b/src/kudu/rpc/negotiation.cc
index b7d7ed31b..77bfa8686 100644
--- a/src/kudu/rpc/negotiation.cc
+++ b/src/kudu/rpc/negotiation.cc
@@ -130,20 +130,19 @@ static Status WaitForClientConnect(Socket* socket, const MonoTime& deadline) {
#endif
if (ready == -1) {
int err = errno;
- if (err == EINTR) {
- // We were interrupted by a signal, let's go again.
- continue;
- } else {
+ if (PREDICT_FALSE(err != EINTR)) {
return Status::NetworkError("Error from ppoll() while waiting to connect",
ErrnoToString(err), err);
}
- } else if (ready == 0) {
+ // We were interrupted by a signal, let's go again.
+ continue;
+ }
+ if (ready == 0) {
// Timeout exceeded. Loop back to the top to our impending doom.
continue;
- } else {
- // Success.
- break;
}
+ // Success.
+ break;
}
// Connect finished, but this doesn't mean that we connected successfully.
@@ -151,7 +150,7 @@ static Status WaitForClientConnect(Socket* socket, const MonoTime& deadline) {
int so_error = 0;
socklen_t socklen = sizeof(so_error);
int rc = getsockopt(fd, SOL_SOCKET, SO_ERROR, &so_error, &socklen);
- if (rc != 0) {
+ if (PREDICT_FALSE(rc != 0)) {
return Status::NetworkError("Unable to check connected socket for errors",
ErrnoToString(errno),
errno);
@@ -262,7 +261,7 @@ static Status DoServerNegotiation(Connection* conn,
messenger->tls_context().is_external_cert())
<< "misconfiguration: check ValidateRpcAuthnFlags() for details";
- if (FLAGS_rpc_negotiation_inject_delay_ms > 0) {
+ if (PREDICT_FALSE(FLAGS_rpc_negotiation_inject_delay_ms > 0)) {
LOG(WARNING) << "Injecting " << FLAGS_rpc_negotiation_inject_delay_ms
<< "ms delay in negotiation";
SleepFor(MonoDelta::FromMilliseconds(FLAGS_rpc_negotiation_inject_delay_ms));
diff --git a/src/kudu/rpc/outbound_call.cc b/src/kudu/rpc/outbound_call.cc
index cd8980798..228a1e091 100644
--- a/src/kudu/rpc/outbound_call.cc
+++ b/src/kudu/rpc/outbound_call.cc
@@ -331,7 +331,7 @@ void OutboundCall::SetResponse(unique_ptr<CallResponse> resp) {
// 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.
- if (!response_->ParseFromArray(r.data(), r.size())) {
+ if (PREDICT_FALSE(!response_->ParseFromArray(r.data(), r.size()))) {
SetFailed(Status::IOError("invalid RPC response, missing fields",
response_->InitializationErrorString()));
return;
@@ -538,7 +538,7 @@ CallResponse::CallResponse()
Status CallResponse::GetSidecar(int idx, Slice* sidecar) const {
DCHECK(parsed_);
- if (idx < 0 || idx >= header_.sidecar_offsets_size()) {
+ if (PREDICT_FALSE(idx < 0 || idx >= header_.sidecar_offsets_size())) {
return Status::InvalidArgument(strings::Substitute(
"Index $0 does not reference a valid sidecar", idx));
}
diff --git a/src/kudu/rpc/proxy.cc b/src/kudu/rpc/proxy.cc
index b24698b67..e90474deb 100644
--- a/src/kudu/rpc/proxy.cc
+++ b/src/kudu/rpc/proxy.cc
@@ -111,7 +111,7 @@ void Proxy::Init(Sockaddr addr) {
// Effective user and password remain blank.
string real_user;
Status s = GetLoggedInUser(&real_user);
- if (!s.ok()) {
+ if (PREDICT_FALSE(!s.ok())) {
LOG(WARNING) << Substitute(
"$0: unable to get logged-in username before connecting to $1 via $2 proxy",
s.ToString(), hp_.ToString(), service_name_);
@@ -172,7 +172,7 @@ void Proxy::RefreshDnsAndEnqueueRequest(const string& method,
unique_ptr<RequestPayload> req_payload(req_raw);
unique_ptr<vector<Sockaddr>> unique_addrs(addrs);
// If we fail to resolve the address, treat the call as failed.
- if (!s.ok() || addrs->empty()) {
+ if (PREDICT_FALSE(!s.ok() || addrs->empty())) {
DCHECK(!controller->call_);
// NOTE: we need to keep a reference here because the callback may end up
// destructing the controller and the outbound call, _while_ the callback
diff --git a/src/kudu/rpc/reactor.cc b/src/kudu/rpc/reactor.cc
index 114a8fba0..eb61c5bf3 100644
--- a/src/kudu/rpc/reactor.cc
+++ b/src/kudu/rpc/reactor.cc
@@ -410,7 +410,7 @@ void ReactorThread::CancelOutboundCall(const shared_ptr<OutboundCall>& call) {
//
void ReactorThread::TimerHandler(ev::timer& /*watcher*/, int revents) {
DCHECK(IsCurrentThread());
- if (EV_ERROR & revents) {
+ if (PREDICT_FALSE(EV_ERROR & revents)) {
LOG(WARNING) << "Reactor " << name() << " got an error in "
"the timer handler.";
return;
@@ -589,7 +589,7 @@ Status ReactorThread::FindOrStartConnection(const ConnectionId& conn_id,
// Kick off blocking client connection negotiation.
Status s = StartConnectionNegotiation(*conn);
- if (s.IsIllegalState()) {
+ if (PREDICT_FALSE(s.IsIllegalState())) {
// Return a nicer error message to the user indicating -- if we just
// forward the status we'd get something generic like "ThreadPool is closing".
return Status::ServiceUnavailable("Client RPC Messenger shutting down");
@@ -764,7 +764,7 @@ void DelayedTask::TimerHandler(ev::timer& /*watcher*/, int revents) {
// We will free this task's memory.
thread_->scheduled_tasks_.erase(thread_->scheduled_tasks_.iterator_to(*this));
- if (EV_ERROR & revents) {
+ if (PREDICT_FALSE(EV_ERROR & revents)) {
string msg = "Delayed task got an error in its timer handler";
LOG(WARNING) << msg;
Abort(Status::Aborted(msg)); // Will delete 'this'.
@@ -971,7 +971,7 @@ void Reactor::ScheduleReactorTask(ReactorTask* task) {
bool Reactor::DrainTaskQueue(boost::intrusive::list<ReactorTask>* tasks) { // NOLINT(*)
std::lock_guard l(lock_);
- if (closing_) {
+ if (PREDICT_FALSE(closing_)) {
return false;
}
tasks->swap(pending_tasks_);
diff --git a/src/kudu/rpc/rpc_controller.cc b/src/kudu/rpc/rpc_controller.cc
index 70dc94686..b362f40b7 100644
--- a/src/kudu/rpc/rpc_controller.cc
+++ b/src/kudu/rpc/rpc_controller.cc
@@ -24,6 +24,7 @@
#include <glog/logging.h>
+#include "kudu/gutil/port.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/rpc/messenger.h"
#include "kudu/rpc/outbound_call.h"
@@ -145,12 +146,12 @@ MonoDelta RpcController::timeout() const {
}
Status RpcController::AddOutboundSidecar(unique_ptr<RpcSidecar> car, int* idx) {
- if (outbound_sidecars_.size() >= TransferLimits::kMaxSidecars) {
+ if (PREDICT_FALSE(outbound_sidecars_.size() >= TransferLimits::kMaxSidecars)) {
return Status::RuntimeError("All available sidecars already used");
}
int64_t sidecar_bytes = car->TotalSize();
- if (outbound_sidecars_total_bytes_ >
- TransferLimits::kMaxTotalSidecarBytes - sidecar_bytes) {
+ if (PREDICT_FALSE(outbound_sidecars_total_bytes_ + sidecar_bytes >
+ TransferLimits::kMaxTotalSidecarBytes)) {
return Status::RuntimeError(Substitute("Total size of sidecars $0 would exceed limit $1",
static_cast<int64_t>(outbound_sidecars_total_bytes_) + sidecar_bytes,
TransferLimits::kMaxTotalSidecarBytes));
diff --git a/src/kudu/rpc/rpc_sidecar.cc b/src/kudu/rpc/rpc_sidecar.cc
index da7244d00..f36ba4fc6 100644
--- a/src/kudu/rpc/rpc_sidecar.cc
+++ b/src/kudu/rpc/rpc_sidecar.cc
@@ -25,6 +25,7 @@
#include <boost/container/vector.hpp>
#include <google/protobuf/repeated_field.h>
+#include "kudu/gutil/port.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/rpc/transfer.h"
#include "kudu/util/faststring.h"
@@ -95,13 +96,13 @@ Status RpcSidecar::ParseSidecars(
if (offsets.empty()) return Status::OK();
int last = offsets.size() - 1;
- if (last >= TransferLimits::kMaxSidecars) {
+ if (PREDICT_FALSE(last >= TransferLimits::kMaxSidecars)) {
return Status::Corruption(strings::Substitute(
"Received $0 additional payload slices, expected at most $1",
last, TransferLimits::kMaxSidecars));
}
- if (buffer.size() > TransferLimits::kMaxTotalSidecarBytes) {
+ if (PREDICT_FALSE(buffer.size() > TransferLimits::kMaxTotalSidecarBytes)) {
return Status::Corruption(strings::Substitute(
"Received $0 payload bytes, expected at most $1",
buffer.size(), TransferLimits::kMaxTotalSidecarBytes));
@@ -111,13 +112,13 @@ Status RpcSidecar::ParseSidecars(
for (int i = 0; i < last; ++i) {
int64_t cur_offset = offsets.Get(i);
int64_t next_offset = offsets.Get(i + 1);
- if (next_offset > buffer.size()) {
+ if (PREDICT_FALSE(next_offset > buffer.size())) {
return Status::Corruption(strings::Substitute(
"Invalid sidecar offsets; sidecar $0 apparently starts at $1,"
" has length $2, but the entire message has length $3",
i, cur_offset, (next_offset - cur_offset), buffer.size()));
}
- if (next_offset < cur_offset) {
+ if (PREDICT_FALSE(next_offset < cur_offset)) {
return Status::Corruption(strings::Substitute(
"Invalid sidecar offsets; sidecar $0 apparently starts at $1,"
" but ends before that at offset $1.", i, cur_offset, next_offset));
@@ -127,7 +128,7 @@ Status RpcSidecar::ParseSidecars(
}
int64_t cur_offset = offsets.Get(last);
- if (cur_offset > buffer.size()) {
+ if (PREDICT_FALSE(cur_offset > buffer.size())) {
return Status::Corruption(strings::Substitute("Invalid sidecar offsets: sidecar $0 "
"starts at offset $1after message ends (message length $2).", last,
cur_offset, buffer.size()));
diff --git a/src/kudu/rpc/serialization.cc b/src/kudu/rpc/serialization.cc
index 40d823ba0..9aa7ab30d 100644
--- a/src/kudu/rpc/serialization.cc
+++ b/src/kudu/rpc/serialization.cc
@@ -126,7 +126,7 @@ Status ParseMessage(const Slice& buf,
DCHECK_EQ(total_len, buf.size() - kMsgLengthPrefixLength)
<< "Got mis-sized buffer: " << KUDU_REDACT(buf.ToDebugString());
- if (total_len > std::numeric_limits<int32_t>::max()) {
+ if (PREDICT_FALSE(total_len > std::numeric_limits<int32_t>::max())) {
return Status::Corruption(Substitute("Invalid packet: message had a length of $0, "
"but we only support messages up to $1 bytes\n",
total_len, std::numeric_limits<int32_t>::max()));
diff --git a/src/kudu/rpc/service_pool.cc b/src/kudu/rpc/service_pool.cc
index ad5fb03cd..c95db25e7 100644
--- a/src/kudu/rpc/service_pool.cc
+++ b/src/kudu/rpc/service_pool.cc
@@ -200,7 +200,7 @@ Status ServicePool::QueueInboundCall(unique_ptr<InboundCall> call) {
void ServicePool::RunThread() {
while (true) {
std::unique_ptr<InboundCall> incoming;
- if (!service_queue_.BlockingGet(&incoming)) {
+ if (PREDICT_FALSE(!service_queue_.BlockingGet(&incoming))) {
VLOG(1) << "ServicePool: messenger shutting down.";
return;
}