This is an automated email from the ASF dual-hosted git repository.
chenBright pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new 881077d1 Support RDMA handshake protocol (#3350)
881077d1 is described below
commit 881077d15bce70df3cec46ed59dfaaf95aa2993b
Author: Bright Chen <[email protected]>
AuthorDate: Sat Jul 18 17:42:52 2026 +0800
Support RDMA handshake protocol (#3350)
* Support RDMA handshake protocol
* Fix comment
---
src/brpc/global.cpp | 10 +
src/brpc/input_messenger.cpp | 11 -
src/brpc/input_messenger.h | 2 +
src/brpc/options.proto | 57 ++---
src/brpc/parse_result.h | 1 +
src/brpc/policy/rdma_handshake_protocol.cpp | 41 ++++
src/brpc/policy/rdma_handshake_protocol.h | 56 +++++
src/brpc/rdma/rdma_endpoint.cpp | 355 ++++++++++++----------------
src/brpc/rdma/rdma_endpoint.h | 22 +-
src/brpc/rdma/rdma_handshake.cpp | 285 +++++++++++++---------
src/brpc/rdma/rdma_handshake.h | 108 ++++-----
src/brpc/rdma/rdma_handshake_constants.h | 56 +++++
src/brpc/rdma/rdma_handshake_server.cpp | 214 +++++++++++++++++
src/brpc/rdma/rdma_handshake_server.h | 48 ++++
src/brpc/rdma_transport.cpp | 21 +-
src/brpc/rdma_transport.h | 1 +
src/brpc/server.cpp | 7 +-
src/brpc/socket.h | 3 +
src/butil/raw_pack.h | 34 ++-
test/brpc_rdma_unittest.cpp | 223 +++++++++--------
20 files changed, 1008 insertions(+), 547 deletions(-)
diff --git a/src/brpc/global.cpp b/src/brpc/global.cpp
index e9402512..04a56282 100644
--- a/src/brpc/global.cpp
+++ b/src/brpc/global.cpp
@@ -69,6 +69,7 @@
// Protocols
#include "brpc/protocol.h"
+#include "brpc/policy/rdma_handshake_protocol.h"
#include "brpc/policy/baidu_rpc_protocol.h"
#include "brpc/policy/http_rpc_protocol.h"
#include "brpc/policy/http2_rpc_protocol.h"
@@ -437,6 +438,15 @@ static void GlobalInitializeOrDieImpl() {
}
// Protocols
+ Protocol rdma_handshake_protocol = {
+ ParseRdmaHandshake, NULL, NULL,
+ ProcessRdmaHandshake, NULL,
+ NULL, NULL, NULL,
+ CONNECTION_TYPE_ALL, "rdma_handshake" };
+ if (RegisterProtocol(PROTOCOL_RDMA_HANDSHAKE, rdma_handshake_protocol) !=
0) {
+ exit(1);
+ }
+
Protocol baidu_protocol = { ParseRpcMessage,
SerializeRpcRequest, PackRpcRequest,
ProcessRpcRequest, ProcessRpcResponse,
diff --git a/src/brpc/input_messenger.cpp b/src/brpc/input_messenger.cpp
index c249cca2..ad47f32e 100644
--- a/src/brpc/input_messenger.cpp
+++ b/src/brpc/input_messenger.cpp
@@ -79,7 +79,6 @@ 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 size_t PROTO_DUMMY_LEN = 4;
ParseResult InputMessenger::CutInputMessage(
Socket* m, size_t* index, bool read_eof) {
@@ -107,16 +106,6 @@ ParseResult InputMessenger::CutInputMessage(
<< " bytes, the connection will be closed."
" Set max_body_size to allow bigger messages";
return result;
- } else {
- if (m->_read_buf.size() >= 4) {
- // The length of `data' must be PROTO_DUMMY_LEN + 1 to
store extra ending char '\0'
- char data[PROTO_DUMMY_LEN + 1];
- m->_read_buf.copy_to_cstr(data, PROTO_DUMMY_LEN);
- if (strncmp(data, "RDMA", PROTO_DUMMY_LEN) == 0) {
- // To avoid timeout when client uses RDMA but server
uses TCP
- return MakeParseError(PARSE_ERROR_TRY_OTHERS);
- }
- }
}
if (m->CreatedByConnect()) {
diff --git a/src/brpc/input_messenger.h b/src/brpc/input_messenger.h
index 8482c3f3..74e40a92 100644
--- a/src/brpc/input_messenger.h
+++ b/src/brpc/input_messenger.h
@@ -30,6 +30,7 @@ namespace rdma {
class RdmaEndpoint;
}
class TcpTransport;
+class RdmaTransport;
struct InputMessageHandler {
// The callback to cut a message from `source'.
// Returned message will be passed to process_request or process_response
@@ -92,6 +93,7 @@ private:
class InputMessenger : public SocketUser {
friend class Socket;
friend class TcpTransport;
+friend class RdmaTransport;
friend class rdma::RdmaEndpoint;
public:
explicit InputMessenger(size_t capacity = 128);
diff --git a/src/brpc/options.proto b/src/brpc/options.proto
index 935caaaa..13b8b682 100644
--- a/src/brpc/options.proto
+++ b/src/brpc/options.proto
@@ -37,35 +37,36 @@ enum ConnectionType {
enum ProtocolType {
PROTOCOL_UNKNOWN = 0;
- PROTOCOL_BAIDU_STD = 1;
- PROTOCOL_STREAMING_RPC = 2;
- PROTOCOL_HULU_PBRPC = 3;
- PROTOCOL_SOFA_PBRPC = 4;
- PROTOCOL_RTMP = 5;
- PROTOCOL_THRIFT = 6;
- PROTOCOL_HTTP = 7;
- PROTOCOL_PUBLIC_PBRPC = 8;
- PROTOCOL_NOVA_PBRPC = 9;
- PROTOCOL_REDIS = 10;
- PROTOCOL_NSHEAD_CLIENT = 11; // implemented in baidu-rpc-ub
- PROTOCOL_NSHEAD = 12;
- PROTOCOL_HADOOP_RPC = 13;
- PROTOCOL_HADOOP_SERVER_RPC = 14;
- PROTOCOL_MONGO = 15; // server side only
- PROTOCOL_UBRPC_COMPACK = 16;
- PROTOCOL_DIDX_CLIENT = 17; // Client side only
- PROTOCOL_MEMCACHE = 18; // Client side only
- PROTOCOL_ITP = 19;
- PROTOCOL_NSHEAD_MCPACK = 20;
- PROTOCOL_DISP_IDL = 21; // Client side only
- PROTOCOL_ERSDA_CLIENT = 22; // Client side only
- PROTOCOL_UBRPC_MCPACK2 = 23; // Client side only
+ PROTOCOL_RDMA_HANDSHAKE = 1;
+ PROTOCOL_BAIDU_STD = 2;
+ PROTOCOL_STREAMING_RPC = 3;
+ PROTOCOL_HULU_PBRPC = 4;
+ PROTOCOL_SOFA_PBRPC = 5;
+ PROTOCOL_RTMP = 6;
+ PROTOCOL_THRIFT = 7;
+ PROTOCOL_HTTP = 8;
+ PROTOCOL_PUBLIC_PBRPC = 9;
+ PROTOCOL_NOVA_PBRPC = 10;
+ PROTOCOL_REDIS = 11;
+ PROTOCOL_NSHEAD_CLIENT = 12; // implemented in baidu-rpc-ub
+ PROTOCOL_NSHEAD = 13;
+ PROTOCOL_HADOOP_RPC = 14;
+ PROTOCOL_HADOOP_SERVER_RPC = 15;
+ PROTOCOL_MONGO = 16; // server side only
+ PROTOCOL_UBRPC_COMPACK = 17;
+ PROTOCOL_DIDX_CLIENT = 18; // Client side only
+ PROTOCOL_MEMCACHE = 19; // Client side only
+ PROTOCOL_ITP = 20;
+ PROTOCOL_NSHEAD_MCPACK = 21;
+ PROTOCOL_DISP_IDL = 22; // Client side only
+ PROTOCOL_ERSDA_CLIENT = 23; // Client side only
+ PROTOCOL_UBRPC_MCPACK2 = 24; // Client side only
// Reserve special protocol for cds-agent, which depends on FIFO right now
- PROTOCOL_CDS_AGENT = 24; // Client side only
- PROTOCOL_ESP = 25; // Client side only
- PROTOCOL_H2 = 26;
- PROTOCOL_COUCHBASE = 27;
- PROTOCOL_MYSQL = 28; // Client side only
+ PROTOCOL_CDS_AGENT = 25; // Client side only
+ PROTOCOL_ESP = 26; // Client side only
+ PROTOCOL_H2 = 27;
+ PROTOCOL_COUCHBASE = 28;
+ PROTOCOL_MYSQL = 29; // Client side only
}
enum CompressType {
diff --git a/src/brpc/parse_result.h b/src/brpc/parse_result.h
index e84f80e0..d916acc5 100644
--- a/src/brpc/parse_result.h
+++ b/src/brpc/parse_result.h
@@ -19,6 +19,7 @@
#ifndef BRPC_PARSE_RESULT_H
#define BRPC_PARSE_RESULT_H
+#include <stddef.h>
namespace brpc {
diff --git a/src/brpc/policy/rdma_handshake_protocol.cpp
b/src/brpc/policy/rdma_handshake_protocol.cpp
new file mode 100644
index 00000000..580abdda
--- /dev/null
+++ b/src/brpc/policy/rdma_handshake_protocol.cpp
@@ -0,0 +1,41 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "brpc/policy/rdma_handshake_protocol.h"
+
+#include "butil/logging.h"
+#include "brpc/destroyable.h"
+#include "brpc/rdma/rdma_handshake_server.h"
+
+namespace brpc {
+namespace policy {
+
+ParseResult ParseRdmaHandshake(butil::IOBuf* source, Socket* socket,
+ bool /*read_eof*/, const void* /*arg*/) {
+ return rdma::ExecuteServerHandshake(source, socket);
+}
+
+void ProcessRdmaHandshake(InputMessageBase* msg) {
+ // ParseRdmaHandshake replies inline and only ever returns
+ // NOT_ENOUGH_DATA / TRY_OTHERS / hard errors, never a real message, so
this
+ // must never run. Keep a placeholder (required for server registration).
+ DestroyingPtr<InputMessageBase> destroying_msg(msg);
+ CHECK(false) << "ProcessRdmaHandshake should never be called";
+}
+
+} // namespace policy
+} // namespace brpc
diff --git a/src/brpc/policy/rdma_handshake_protocol.h
b/src/brpc/policy/rdma_handshake_protocol.h
new file mode 100644
index 00000000..e569a4c3
--- /dev/null
+++ b/src/brpc/policy/rdma_handshake_protocol.h
@@ -0,0 +1,56 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#ifndef BRPC_POLICY_RDMA_HANDSHAKE_PROTOCOL_H
+#define BRPC_POLICY_RDMA_HANDSHAKE_PROTOCOL_H
+
+// NOTE: This file is intentionally INDEPENDENT of BRPC_WITH_RDMA. A server may
+// run in TCP mode either because it was built without RDMA, or because RDMA
was
+// not enabled at runtime. In both cases an RDMA client that connects to it
will
+// send an RDMA handshake magic ("RDMA" for v2, "RDM3" for v3) first. Without
+// special handling the server treats those bytes as an unknown protocol and
+// closes the connection, so the client (blocked reading the server hello) only
+// sees EOF and cannot fall back to TCP.
+//
+// To let the client fall back on the SAME connection, the server recognizes
the
+// RDMA handshake as a "first-class" protocol (magic in the first 4 bytes,
+// PROTOCOL_RDMA_HANDSHAKE is ordered before PROTOCOL_HTTP), replies a hello
with
+// an incompatible version so the client rejects it and downgrades to TCP, then
+// drains the client's subsequent ACK and lets normal RPC parsing continue.
+
+#include "butil/iobuf.h"
+#include "brpc/input_message_base.h"
+#include "brpc/parse_result.h"
+#include "brpc/socket.h"
+
+namespace brpc {
+namespace policy {
+
+// Parse binary format of rdma handshake.
+ParseResult ParseRdmaHandshake(butil::IOBuf* source, Socket* socket,
+ bool read_eof, const void* arg);
+
+// Actions to a rdma handshake request, which is left unimplemented.
+// All requests are processed in the parsing process. This function
+// must be declared since server only enables rdma handshake as a
+// server-side protocol when this function is declared.
+void ProcessRdmaHandshake(InputMessageBase* msg);
+
+} // namespace policy
+} // namespace brpc
+
+#endif // BRPC_POLICY_RDMA_HANDSHAKE_PROTOCOL_H
diff --git a/src/brpc/rdma/rdma_endpoint.cpp b/src/brpc/rdma/rdma_endpoint.cpp
index a5016dc7..daefbcb9 100644
--- a/src/brpc/rdma/rdma_endpoint.cpp
+++ b/src/brpc/rdma/rdma_endpoint.cpp
@@ -32,6 +32,7 @@
#include "brpc/rdma/rdma_endpoint.h"
#include "brpc/rdma_transport.h"
#include "brpc/rdma/rdma_handshake.h"
+#include "brpc/rdma/rdma_handshake_constants.h"
DECLARE_int32(task_group_ntags);
@@ -84,14 +85,6 @@ extern const uint16_t MIN_QP_SIZE = 16;
static const uint16_t MAX_QP_SIZE = 4096;
extern const uint16_t MIN_BLOCK_SIZE = 1024;
-// ACK message wire format (shared by all protocol versions): a single
-// 4B big-endian flags word; bit 0 (HELLO_ACK_RDMA_OK) indicates the
-// sender wants to use RDMA. The state machines in
-// ProcessHandshakeAt{Client,Server} inline the corresponding 4B
-// send/recv directly using ReadFromFd / WriteToFd.
-static const size_t HELLO_ACK_LEN = 4;
-static const uint32_t HELLO_ACK_RDMA_OK = 0x1;
-
static butil::Mutex* g_rdma_resource_mutex = NULL;
static RdmaResource* g_rdma_resource_list = NULL;
@@ -161,7 +154,7 @@ RdmaEndpoint::~RdmaEndpoint() {
void RdmaEndpoint::Reset() {
DeallocateResources();
- _state.store(UNINIT, butil::memory_order_relaxed);
+ _state = UNINIT;
_resource = NULL;
_send_cq_events = 0;
_recv_cq_events = 0;
@@ -195,8 +188,7 @@ void RdmaConnect::StartConnect(const Socket* socket,
return;
}
if (!IsRdmaAvailable()) {
- rdma_transport->_rdma_ep->_state.store(RdmaEndpoint::FALLBACK_TCP,
- butil::memory_order_relaxed);
+ rdma_transport->_rdma_ep->_state = RdmaEndpoint::FALLBACK_TCP;
rdma_transport->_rdma_state = RdmaTransport::RDMA_OFF;
done(0, data);
return;
@@ -222,34 +214,6 @@ void RdmaConnect::Run() {
_done(errno, _data);
}
-static void TryReadOnTcpDuringRdmaEst(Socket* s) {
- int progress = Socket::PROGRESS_INIT;
- while (true) {
- uint8_t tmp;
- ssize_t nr = read(s->fd(), &tmp, 1);
- if (nr < 0) {
- if (errno != EAGAIN) {
- const int saved_errno = errno;
- PLOG(WARNING) << "Fail to read from " << s;
- s->SetFailed(saved_errno, "Fail to read from %s: %s",
- s->description().c_str(), berror(saved_errno));
- return;
- }
- if (!s->MoreReadEvents(&progress)) {
- break;
- }
- } else if (nr == 0) {
- s->SetEOF();
- return;
- } else {
- LOG(WARNING) << "Read unexpected data from " << s;
- s->SetFailed(EPROTO, "Read unexpected data from %s",
- s->description().c_str());
- return;
- }
- }
-}
-
void RdmaEndpoint::OnNewDataFromTcp(Socket* m) {
auto* rdma_transport = static_cast<RdmaTransport*>(m->_transport.get());
RdmaEndpoint* ep = rdma_transport->GetRdmaEp();
@@ -257,40 +221,36 @@ void RdmaEndpoint::OnNewDataFromTcp(Socket* m) {
int progress = Socket::PROGRESS_INIT;
while (true) {
- State state = ep->_state.load(butil::memory_order_acquire);
- if (state == UNINIT) {
- if (!m->CreatedByConnect()) {
- if (!IsRdmaAvailable()) {
- rdma_transport->_rdma_state = RdmaTransport::RDMA_OFF;
- ep->_state.store(FALLBACK_TCP,
butil::memory_order_relaxed);
- continue;
- }
- bthread_t tid;
- ep->_state.store(S_HELLO_WAIT, butil::memory_order_relaxed);
- SocketUniquePtr s;
- m->ReAddress(&s);
- bthread_attr_t attr = BTHREAD_ATTR_NORMAL;
- bthread_attr_set_name(&attr, "RdmaProcessHandshakeAtServer");
- if (bthread_start_background(&tid, &attr,
ProcessHandshakeAtServer, ep) < 0) {
- ep->_state.store(UNINIT, butil::memory_order_relaxed);
- LOG(FATAL) << "Fail to start handshake bthread";
- } else {
- s.release();
- }
- } else {
- // The connection may be closed or reset before the client
- // starts handshake. This will be handled by client handshake.
- // Ignore the exception here.
- }
- } else if (state < ESTABLISHED) { // during handshake
+ if (ep->_state == UNINIT) {
+ // The connection may be closed or reset before the client starts
+ // handshake. This will be handled by client handshake. Ignore
here.
+ } else if (ep->_state < ESTABLISHED) { // during handshake
ep->_read_butex->fetch_add(1, butil::memory_order_release);
bthread::butex_wake(ep->_read_butex);
- } else if (state == FALLBACK_TCP){ // handshake finishes
+ } else if (ep->_state == FALLBACK_TCP){ // handshake finishes
InputMessenger::OnNewMessages(m);
return;
- } else if (state == ESTABLISHED) {
- TryReadOnTcpDuringRdmaEst(ep->_socket);
- return;
+ } else if (ep->_state == ESTABLISHED) {
+ uint8_t tmp;
+ ssize_t nr = read(ep->_socket->fd(), &tmp, 1);
+ if (nr == 0) {
+ ep->_socket->SetEOF();
+ return;
+ }
+ if (nr > 0) {
+ LOG(WARNING) << "Read unexpected data from " << ep->_socket;
+ ep->_socket->SetFailed(EPROTO, "Read unexpected data from %s",
+ ep->_socket->description().c_str());
+ return;
+ }
+
+ if (errno != EAGAIN) {
+ const int saved_errno = errno;
+ PLOG(WARNING) << "Fail to read from " << ep->_socket;
+ ep->_socket->SetFailed(saved_errno, "Fail to read from %s: %s",
+ ep->_socket->description().c_str(),
+ berror(saved_errno));
+ }
}
if (!m->MoreReadEvents(&progress)) {
break;
@@ -422,28 +382,13 @@ int RdmaEndpoint::WriteToFd(butil::IOBuf* data) {
});
}
-inline void RdmaEndpoint::TryReadOnTcp() {
- if (_socket->_nevent.fetch_add(1, butil::memory_order_acq_rel) == 0) {
- State state = _state.load(butil::memory_order_acquire);
- if (state == FALLBACK_TCP) {
- InputMessenger::OnNewMessages(_socket);
- } else if (state == ESTABLISHED) {
- TryReadOnTcpDuringRdmaEst(_socket);
- }
- }
-}
-
void RdmaEndpoint::ApplyRemoteHello(const ParsedHello& remote) {
_remote_recv_block_size = remote.block_size;
- _local_window_capacity =
- std::min(_sq_size, remote.rq_size) - RESERVED_WR_NUM;
- _remote_window_capacity =
- std::min(_rq_size, remote.sq_size) - RESERVED_WR_NUM;
+ _local_window_capacity = std::min(_sq_size, remote.rq_size) -
RESERVED_WR_NUM;
+ _remote_window_capacity = std::min(_rq_size, remote.sq_size) -
RESERVED_WR_NUM;
_sq_imm_window_size = RESERVED_WR_NUM;
- _remote_rq_window_size.store(
- _local_window_capacity, butil::memory_order_relaxed);
- _sq_window_size.store(
- _local_window_capacity, butil::memory_order_relaxed);
+ _remote_rq_window_size.store(_local_window_capacity,
butil::memory_order_relaxed);
+ _sq_window_size.store(_local_window_capacity, butil::memory_order_relaxed);
}
// Client-side handshake entry: the state machine.
@@ -478,47 +423,47 @@ void* RdmaEndpoint::ProcessHandshakeAtClient(void* arg) {
ep->_handshake_version = handshake->ProtocolVersion();
// First initialize CQ and QP resources.
- ep->_state.store(C_ALLOC_QPCQ, butil::memory_order_relaxed);
+ ep->_state = C_ALLOC_QPCQ;
if (ep->AllocateResources() < 0) {
LOG(WARNING) << "Fallback to tcp:" << s->description();
rdma_transport->_rdma_state = RdmaTransport::RDMA_OFF;
- ep->_state.store(FALLBACK_TCP, butil::memory_order_release);
+ ep->_state = FALLBACK_TCP;
return NULL;
}
// Send hello message to server
- ep->_state.store(C_HELLO_SEND, butil::memory_order_relaxed);
+ ep->_state = C_HELLO_SEND;
if (handshake->SendLocalHello() < 0) {
int saved_errno = errno;
PLOG(WARNING) << "Fail to send hello message to server:"
<< s->description();
s->SetFailed(saved_errno, "Fail to complete rdma handshake from %s:
%s",
s->description().c_str(), berror(saved_errno));
- ep->_state.store(FAILED, butil::memory_order_relaxed);
+ ep->_state = FAILED;
return NULL;
}
// Receive and parse remote hello.
- ep->_state.store(C_HELLO_WAIT, butil::memory_order_relaxed);
+ ep->_state = C_HELLO_WAIT;
ParsedHello remote{};
- bool negotiated = false;
- if (handshake->ReceiveAndParseRemoteHello(&remote, &negotiated) < 0) {
+ const RemoteHelloResult r = handshake->ReceiveAndParseRemoteHello(&remote);
+ if (r == RemoteHelloResult::ERROR) {
int saved_errno = errno;
PLOG(WARNING) << "Fail to receive hello from server:"
<< s->description();
s->SetFailed(saved_errno, "Fail to complete rdma handshake from %s:
%s",
s->description().c_str(), berror(saved_errno));
- ep->_state.store(FAILED, butil::memory_order_relaxed);
+ ep->_state = FAILED;
return NULL;
}
- if (!negotiated) {
+ if (r != RemoteHelloResult::NEGOTIATED) {
LOG(WARNING) << "Fail to negotiate with server, fallback to tcp:"
<< s->description();
rdma_transport->_rdma_state = RdmaTransport::RDMA_OFF;
} else {
ep->ApplyRemoteHello(remote);
- ep->_state.store(C_BRINGUP_QP, butil::memory_order_relaxed);
+ ep->_state = C_BRINGUP_QP;
if (ep->BringUpQp(remote.lid, remote.gid, remote.qp_num) < 0) {
LOG(WARNING) << "Fail to bringup QP, fallback to tcp:"
<< s->description();
@@ -529,7 +474,7 @@ void* RdmaEndpoint::ProcessHandshakeAtClient(void* arg) {
}
// Send ACK message to server
- ep->_state.store(C_ACK_SEND, butil::memory_order_relaxed);
+ ep->_state = C_ACK_SEND;
bool rdma_on = rdma_transport->_rdma_state == RdmaTransport::RDMA_ON;
uint32_t flags = rdma_on ? HELLO_ACK_RDMA_OK : 0;
uint32_t flags_be = butil::HostToNet32(flags);
@@ -539,17 +484,17 @@ void* RdmaEndpoint::ProcessHandshakeAtClient(void* arg) {
<< s->description();
s->SetFailed(saved_errno, "Fail to complete rdma handshake from %s:
%s",
s->description().c_str(), berror(saved_errno));
- ep->_state.store(FAILED, butil::memory_order_relaxed);
+ ep->_state = FAILED;
return NULL;
}
if (rdma_transport->_rdma_state == RdmaTransport::RDMA_ON) {
- ep->_state.store(ESTABLISHED, butil::memory_order_release);
+ ep->_state = ESTABLISHED;
LOG_IF(INFO, FLAGS_rdma_trace_verbose)
<< "Client handshake ends (use rdma v" << ep->_handshake_version
<< ") on " << s->description();
} else {
- ep->_state.store(FALLBACK_TCP, butil::memory_order_release);
+ ep->_state = FALLBACK_TCP;
LOG_IF(INFO, FLAGS_rdma_trace_verbose)
<< "Client handshake ends (use tcp) on " << s->description();
}
@@ -574,129 +519,119 @@ void* RdmaEndpoint::ProcessHandshakeAtClient(void* arg)
{
// |
// v
// ESTABLISHED / FALLBACK_TCP
-void* RdmaEndpoint::ProcessHandshakeAtServer(void* arg) {
- auto ep = static_cast<RdmaEndpoint*>(arg);
- SocketUniquePtr s(ep->_socket);
- auto rdma_transport = static_cast<RdmaTransport*>(s->_transport.get());
+ParseResult RdmaEndpoint::ExecuteServerHandshake(butil::IOBuf* source, Socket*
s) {
+ RdmaTransport* rdma_transport =
static_cast<RdmaTransport*>(s->_transport.get());
+ RdmaEndpoint* ep = rdma_transport->_rdma_ep;
+ CHECK(ep != NULL);
- LOG_IF(INFO, FLAGS_rdma_trace_verbose)
- << "Start handshake on " << s->description();
+ if (s->parsing_context() == NULL) {
+ // Phase 1: read the client hello, negotiate, reply server hello.
+ if (source->size() < HELLO_MAGIC_LEN) {
+ return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
+ }
+ uint8_t magic[HELLO_MAGIC_LEN];
+ CHECK_EQ(source->copy_to(magic, HELLO_MAGIC_LEN), HELLO_MAGIC_LEN);
+
+ // Pick the version-specific server handshake from the peeked magic
(the
+ // magic is NOT consumed; ReceiveAndParseRemoteHello() reads it again
+ // from `source`).
+ std::unique_ptr<RdmaHandshake> hs = CreateServerHandshakeByMagic(ep,
source, magic);
+ if (hs == NULL) {
+ return MakeParseError(PARSE_ERROR_TRY_OTHERS);
+ }
+ ep->_handshake_version = hs->ProtocolVersion();
+ ep->_state = S_HELLO_WAIT;
+
+ ParsedHello remote{};
+ const RemoteHelloResult r = hs->ReceiveAndParseRemoteHello(&remote);
+ if (r == RemoteHelloResult::NEED_MORE) {
+ return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
+ }
+ if (r == RemoteHelloResult::ERROR) {
+ ep->_state = FAILED;
+ return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
+ }
+
+ // Negotiate + allocate resources.
+ bool negotiated = r == RemoteHelloResult::NEGOTIATED;
+ if (negotiated) {
+ ep->ApplyRemoteHello(remote);
+ ep->_state = S_ALLOC_QPCQ;
+ if (ep->AllocateResources() < 0) {
+ LOG(WARNING) << "Fail to allocate rdma resources, fallback to
tcp:"
+ << s->description();
+ negotiated = false;
+ } else {
+ ep->_state = S_BRINGUP_QP;
+ if (ep->BringUpQp(remote.lid, remote.gid, remote.qp_num) < 0) {
+ LOG(WARNING) << "Fail to bringup QP, fallback to tcp:"
+ << s->description();
+ negotiated = false;
+ }
+ }
+ }
+ if (!negotiated) {
+ rdma_transport->_rdma_state = RdmaTransport::RDMA_OFF;
+ }
- ep->_state.store(S_HELLO_WAIT, butil::memory_order_relaxed);
- uint8_t magic[MAGIC_STR_LEN];
- if (ep->ReadFromFd(magic, MAGIC_STR_LEN) < 0) {
- int saved_errno = errno;
- PLOG(WARNING) << "Fail to read Hello Message from client:"
- << s->description() << " " << s->_remote_side;
- s->SetFailed(saved_errno, "Fail to complete rdma handshake from %s:
%s",
- s->description().c_str(), berror(saved_errno));
- ep->_state.store(FAILED, butil::memory_order_relaxed);
- return NULL;
- }
+ // Reply the server hello.
+ // Emits a real hello when _rdma_state != RDMA_OFF;
+ // an un-negotiable one otherwise.
+ ep->_state = S_HELLO_SEND;
+ if (hs->SendLocalHello() < 0) {
+ PLOG(WARNING) << "Fail to send server hello to " <<
s->description();
+ ep->_state = FAILED;
+ return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
+ }
- // Dispatch on magic, or fall back to TCP
- std::unique_ptr<RdmaHandshake> handshake =
CreateServerHandshakeByMagic(ep, magic);
- if (!handshake) {
- LOG_IF(INFO, FLAGS_rdma_trace_verbose)
- << "It seems that the client does not use RDMA, fallback to TCP:"
- << s->description();
- // We need to copy data read back to _socket->_read_buf.
- s->_read_buf.append(magic, MAGIC_STR_LEN);
- rdma_transport->_rdma_state = RdmaTransport::RDMA_OFF;
- // Use release memory order to publish the magic bytes appended
- // above to whoever reads `_state == FALLBACK_TCP` (the event
- // thread in OnNewDataFromTcp).
- ep->_state.store(FALLBACK_TCP, butil::memory_order_release);
- ep->TryReadOnTcp();
- return NULL;
+ // Enter the wait-ACK phase. Whether negotiation succeeded is already
+ // recorded in rdma_transport->_rdma_state (RDMA_OFF iff negotiation
+ // failed), so the context itself needs no extra flag.
+ s->reset_parsing_context(ServerHandshakeContext::Create());
+ ep->_state = S_ACK_WAIT;
+ return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
}
- ep->_handshake_version = handshake->ProtocolVersion();
- // Magic was already consumed above; the subclass MUST NOT re-read it.
- ParsedHello remote{};
- bool negotiated = false;
- if (handshake->ReceiveAndParseRemoteHello(&remote, &negotiated) < 0) {
- int saved_errno = errno;
- PLOG(WARNING) << "Fail to receive hello from client:"
- << s->description();
- s->SetFailed(saved_errno, "Fail to complete rdma handshake from %s:
%s",
- s->description().c_str(), berror(saved_errno));
- ep->_state.store(FAILED, butil::memory_order_relaxed);
- return NULL;
+ // Phase 2: drain the 4B ACK and finalize.
+ if (source->size() < HELLO_ACK_LEN) {
+ return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
}
-
- if (!negotiated) {
- LOG(WARNING) << "Fail to negotiate with client, fallback to tcp:"
+ if (source->size() > HELLO_ACK_LEN) {
+ LOG(WARNING) << "Too many bytes in handshake ACK, drop connection: "
<< s->description();
- rdma_transport->_rdma_state = RdmaTransport::RDMA_OFF;
- } else {
- ep->ApplyRemoteHello(remote);
- ep->_state.store(S_ALLOC_QPCQ, butil::memory_order_relaxed);
- if (ep->AllocateResources() < 0) {
- LOG(WARNING) << "Fail to allocate rdma resources, fallback to tcp:"
- << s->description();
- rdma_transport->_rdma_state = RdmaTransport::RDMA_OFF;
- } else {
- ep->_state.store(S_BRINGUP_QP, butil::memory_order_relaxed);
- if (ep->BringUpQp(remote.lid, remote.gid, remote.qp_num) < 0) {
- LOG(WARNING) << "Fail to bringup QP, fallback to tcp:"
- << s->description();
- rdma_transport->_rdma_state = RdmaTransport::RDMA_OFF;
- }
- }
+ ep->_state = FAILED;
+ s->reset_parsing_context(NULL);
+ return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
}
- ep->_state.store(S_HELLO_SEND, butil::memory_order_relaxed);
- if (handshake->SendLocalHello() < 0) {
- int saved_errno = errno;
- PLOG(WARNING) << "Fail to send Hello Message to client:"
- << s->description();
- s->SetFailed(saved_errno, "Fail to complete rdma handshake from %s:
%s",
- s->description().c_str(), berror(saved_errno));
- ep->_state.store(FAILED, butil::memory_order_relaxed);
- return NULL;
- }
-
- ep->_state.store(S_ACK_WAIT, butil::memory_order_relaxed);
uint32_t flags_be = 0;
- if (ep->ReadFromFd(&flags_be, HELLO_ACK_LEN) < 0) {
- int saved_errno = errno;
- PLOG(WARNING) << "Fail to read ack message from client:"
- << s->description();
- s->SetFailed(saved_errno, "Fail to complete rdma handshake from %s:
%s",
- s->description().c_str(), berror(saved_errno));
- ep->_state.store(FAILED, butil::memory_order_relaxed);
- return NULL;
- }
+ CHECK_EQ(source->cutn(&flags_be, HELLO_ACK_LEN), HELLO_ACK_LEN);
uint32_t flags = butil::NetToHost32(flags_be);
bool client_ack_ok = (flags & HELLO_ACK_RDMA_OK) != 0;
- if (client_ack_ok) {
- if (rdma_transport->_rdma_state == RdmaTransport::RDMA_OFF) {
- // Client asked for RDMA but we are falling back: protocol
- // breakdown, abort the connection so the client sees a
- // clean error rather than a half-up RDMA channel.
- LOG(WARNING) << "Client wants RDMA in ACK but server is in "
- << "RDMA_OFF state: " << s->description();
- s->SetFailed(EPROTO, "Fail to complete rdma handshake from %s: %s",
- s->description().c_str(), berror(EPROTO));
- ep->_state.store(FAILED, butil::memory_order_relaxed);
- return NULL;
- }
- rdma_transport->_rdma_state = RdmaTransport::RDMA_ON;
- ep->_state.store(ESTABLISHED, butil::memory_order_release);
- LOG_IF(INFO, FLAGS_rdma_trace_verbose)
- << "Server handshake ends (use rdma v" << ep->_handshake_version
- << ") on " << s->description();
- } else {
- rdma_transport->_rdma_state = RdmaTransport::RDMA_OFF;
- ep->_state.store(FALLBACK_TCP, butil::memory_order_release);
+ if (!client_ack_ok) {
LOG_IF(INFO, FLAGS_rdma_trace_verbose)
<< "Server handshake ends (use tcp) on " << s->description();
+ rdma_transport->_rdma_state = RdmaTransport::RDMA_OFF;
+ ep->_state = FALLBACK_TCP;
+ s->reset_parsing_context(NULL);
+ return MakeParseError(PARSE_ERROR_TRY_OTHERS);
}
- ep->TryReadOnTcp();
+ if (rdma_transport->_rdma_state == RdmaTransport::RDMA_OFF) {
+ LOG(WARNING) << "Client wants RDMA in ACK but server fell back: "
+ << s->description();
+ ep->_state = FAILED;
+ s->reset_parsing_context(NULL);
+ return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
+ }
- return NULL;
+ LOG_IF(INFO, FLAGS_rdma_trace_verbose)
+ << "Server handshake ends (use rdma v" << ep->_handshake_version
+ << ") on " << s->description();
+ rdma_transport->_rdma_state = RdmaTransport::RDMA_ON;
+ ep->_state = ESTABLISHED;
+ s->reset_parsing_context(NULL);
+ return MakeParseError(PARSE_ERROR_TRY_OTHERS);
}
bool RdmaEndpoint::IsWritable() const {
@@ -974,7 +909,7 @@ ssize_t RdmaEndpoint::HandleCompletion(ibv_wc& wc) {
if (wc.byte_len < (uint32_t)FLAGS_rdma_zerocopy_min_size) {
zerocopy = false;
}
- CHECK_NE(_state.load(butil::memory_order_acquire), FALLBACK_TCP);
+ CHECK_NE(_state, FALLBACK_TCP);
if (zerocopy) {
_rbuf[_rq_received].cutn(&_socket->_read_buf, wc.byte_len);
} else {
@@ -1593,7 +1528,7 @@ void RdmaEndpoint::PollCq(Socket* m) {
}
std::string RdmaEndpoint::GetStateStr() const {
- switch (_state.load(butil::memory_order_relaxed)) {
+ switch (_state) {
case UNINIT: return "UNINIT";
case C_ALLOC_QPCQ: return "C_ALLOC_QPCQ";
case C_HELLO_SEND: return "C_HELLO_SEND";
diff --git a/src/brpc/rdma/rdma_endpoint.h b/src/brpc/rdma/rdma_endpoint.h
index 41c33824..49ee2c9e 100644
--- a/src/brpc/rdma/rdma_endpoint.h
+++ b/src/brpc/rdma/rdma_endpoint.h
@@ -30,6 +30,7 @@
#include "butil/macros.h"
#include "butil/containers/mpsc_queue.h"
#include "brpc/socket.h"
+#include "brpc/rdma/rdma_handshake_server.h"
namespace brpc {
@@ -45,10 +46,11 @@ class RdmaHandshakeServerV2;
class RdmaHandshakeClientV3;
class RdmaHandshakeServerV3;
struct ParsedHello;
+enum class RemoteHelloResult;
class RdmaHello;
class RdmaEndpoint;
namespace v2_wire {
- int ReadBodyAndNegotiate(RdmaEndpoint* ep, ParsedHello* remote, bool*
negotiated);
+ RemoteHelloResult ReadBodyAndNegotiate(RdmaEndpoint* ep, ParsedHello*
remote);
int DrainBytes(RdmaEndpoint* ep, size_t n);
} // namespace v2_wire
@@ -96,7 +98,7 @@ friend class RdmaHandshakeClientV2;
friend class RdmaHandshakeServerV2;
friend class RdmaHandshakeClientV3;
friend class RdmaHandshakeServerV3;
-friend int v2_wire::ReadBodyAndNegotiate(RdmaEndpoint*, ParsedHello*, bool*);
+friend RemoteHelloResult v2_wire::ReadBodyAndNegotiate(RdmaEndpoint*,
ParsedHello*);
friend int v2_wire::DrainBytes(RdmaEndpoint*, size_t);
friend void v3_wire::FillLocalRdmaHello(const RdmaEndpoint*, RdmaHello*);
friend int v3_wire::ReadAndParseV3Hello(RdmaEndpoint*, RdmaHello*);
@@ -125,9 +127,13 @@ public:
void DebugInfo(std::ostream& os,
butil::StringPiece connector = "\n") const;
- // Callback when there is new epollin event on TCP fd
+ // Callback when there is new epollin event on TCP fd.
+ // Only used by client-side RDMA sockets.
static void OnNewDataFromTcp(Socket* m);
+ // Real handshake for RDMA-mode sockets.
+ static ParseResult ExecuteServerHandshake(butil::IOBuf* source, Socket*
socket);
+
// Initialize polling mode
static int PollingModeInitialize(bthread_tag_t tag,
std::function<void(void)> callback,
@@ -157,9 +163,6 @@ private:
// Process handshake at the client
static void* ProcessHandshakeAtClient(void* arg);
- // Process handshake at the server
- static void* ProcessHandshakeAtServer(void* arg);
-
// Allocate resources
// Return 0 if success, -1 if failed and errno set
int AllocateResources();
@@ -249,9 +252,6 @@ private:
// Get the description of current handshake state
std::string GetStateStr() const;
- // Try to read data on TCP fd in _socket
- void TryReadOnTcp();
-
// Add cq socket id to poller
void PollerAddCqSid();
@@ -261,8 +261,8 @@ private:
// Not owner
Socket* _socket;
- // State of Handshake
- butil::atomic<State> _state;
+ // State of Handshake.
+ State _state;
// Wire-level handshake protocol version (set by dispatch in
// ProcessHandshakeAtClient/Server). Aligned with the protocol code:
diff --git a/src/brpc/rdma/rdma_handshake.cpp b/src/brpc/rdma/rdma_handshake.cpp
index 9bd2312e..1e6004ff 100644
--- a/src/brpc/rdma/rdma_handshake.cpp
+++ b/src/brpc/rdma/rdma_handshake.cpp
@@ -18,6 +18,7 @@
#if BRPC_WITH_RDMA
#include "brpc/rdma/rdma_handshake.h"
+#include "brpc/rdma/rdma_handshake_constants.h"
#include <string.h>
#include <algorithm> // std::min
@@ -26,11 +27,13 @@
#include <gflags/gflags.h>
#include "butil/iobuf.h" // IOBuf, IOPortal, IOBufAsZeroCopy*Stream
#include "butil/sys_byteorder.h"
+#include "butil/raw_pack.h" // RawPacker, RawUnpacker
#include "brpc/socket.h"
#include "brpc/rdma/rdma_endpoint.h"
#include "brpc/rdma/rdma_helper.h"
#include "brpc/rdma_transport.h"
#include "brpc/rdma/rdma_handshake.pb.h"
+#include "butil/object_pool.h"
namespace brpc {
namespace rdma {
@@ -40,58 +43,46 @@ DEFINE_int32(rdma_client_handshake_version, 2,
"2 = legacy 'RDMA' magic (default, compatible with all servers); "
"3 = new 'RDM3' protobuf-based handshake "
"(MUST only be enabled after target servers support v3).");
+DECLARE_bool(rdma_trace_verbose);
extern const uint16_t MIN_QP_SIZE;
extern const uint16_t MIN_BLOCK_SIZE;
extern uint32_t g_rdma_recv_block_size;
extern bool g_skip_rdma_init;
-// Wire-level constants for the v2 handshake.
-static const char* MAGIC_STR = "RDMA";
-static constexpr uint16_t RDMA_HELLO_V2_MSG_LEN = 40; // In Byte
-extern const uint16_t RDMA_HELLO_V2_VERSION = 2;
-extern const uint16_t RDMA_IMPL_V2_VERSION = 1;
-
-// Wire-level constants for the v3 handshake.
-static const char* MAGIC_STR_V3 = "RDM3";
-static const size_t RDMA_HELLO_V3_PB_SIZE_LEN = 4;
-static const size_t RDMA_HELLO_V3_MAX_PB_SIZE = 4096;
-
namespace v2_wire {
void HelloMessage::Serialize(void* data) const {
- uint16_t* current_pos = (uint16_t*)data;
- *(current_pos++) = butil::HostToNet16(msg_len);
- *(current_pos++) = butil::HostToNet16(hello_ver);
- *(current_pos++) = butil::HostToNet16(impl_ver);
- uint32_t* block_size_pos = (uint32_t*)current_pos;
- *block_size_pos = butil::HostToNet32(block_size);
- current_pos += 2; // move forward 4 Bytes
- *(current_pos++) = butil::HostToNet16(sq_size);
- *(current_pos++) = butil::HostToNet16(rq_size);
- *(current_pos++) = butil::HostToNet16(lid);
- fast_memcpy(current_pos, gid.raw, 16);
- uint32_t* qp_num_pos = (uint32_t*)((char*)current_pos + 16);
- *qp_num_pos = butil::HostToNet32(qp_num);
+ butil::RawPacker(data)
+ .pack16(msg_len)
+ .pack16(hello_ver)
+ .pack16(impl_ver)
+ .pack32(block_size)
+ .pack16(sq_size)
+ .pack16(rq_size)
+ .pack16(lid)
+ // gid is a raw 16-byte identifier and must NOT be byte-swapped.
+ .pack_bytes(gid.raw, sizeof(gid.raw))
+ .pack32(qp_num);
}
void HelloMessage::Deserialize(void* data) {
- uint16_t* current_pos = (uint16_t*)data;
- msg_len = butil::NetToHost16(*current_pos++);
- hello_ver = butil::NetToHost16(*current_pos++);
- impl_ver = butil::NetToHost16(*current_pos++);
- block_size = butil::NetToHost32(*(uint32_t*)current_pos);
- current_pos += 2; // move forward 4 Bytes
- sq_size = butil::NetToHost16(*current_pos++);
- rq_size = butil::NetToHost16(*current_pos++);
- lid = butil::NetToHost16(*current_pos++);
- fast_memcpy(gid.raw, current_pos, 16);
- qp_num = butil::NetToHost32(*(uint32_t*)((char*)current_pos + 16));
+ butil::RawUnpacker(data)
+ .unpack16(msg_len)
+ .unpack16(hello_ver)
+ .unpack16(impl_ver)
+ .unpack32(block_size)
+ .unpack16(sq_size)
+ .unpack16(rq_size)
+ .unpack16(lid)
+ // gid is a raw 16-byte identifier and must NOT be byte-swapped.
+ .unpack_bytes(gid.raw, sizeof(gid.raw))
+ .unpack32(qp_num);
}
static bool ValidHelloMessage(const HelloMessage& msg) {
- return msg.hello_ver == RDMA_HELLO_V2_VERSION &&
- msg.impl_ver == RDMA_IMPL_V2_VERSION &&
+ return msg.hello_ver == HELLO_V2_VERSION &&
+ msg.impl_ver == IMPL_V2_VERSION &&
msg.block_size >= MIN_BLOCK_SIZE &&
msg.sq_size >= MIN_QP_SIZE &&
msg.rq_size >= MIN_QP_SIZE;
@@ -106,35 +97,33 @@ static void TranslateV2Hello(const HelloMessage& msg,
ParsedHello* out) {
out->qp_num = msg.qp_num;
}
-int ReadBodyAndNegotiate(RdmaEndpoint* ep, ParsedHello* remote, bool*
negotiated) {
- uint8_t data[HELLO_MSG_LEN_MIN];
- if (ep->ReadFromFd(data, HELLO_MSG_LEN_MIN - MAGIC_STR_LEN) < 0) {
- return -1;
+RemoteHelloResult ReadBodyAndNegotiate(RdmaEndpoint* ep, ParsedHello* remote) {
+ uint8_t data[HELLO_V2_MSG_LEN_MIN];
+ if (ep->ReadFromFd(data, HELLO_V2_MSG_LEN_MIN - HELLO_MAGIC_LEN) < 0) {
+ return RemoteHelloResult::ERROR;
}
HelloMessage remote_msg{};
remote_msg.Deserialize(data);
- if (remote_msg.msg_len < HELLO_MSG_LEN_MIN ||
- remote_msg.msg_len > HELLO_MSG_LEN_MAX) {
+ if (remote_msg.msg_len < HELLO_V2_MSG_LEN_MIN ||
+ remote_msg.msg_len > HELLO_V2_MSG_LEN_MAX) {
errno = EPROTO;
- return -1;
+ return RemoteHelloResult::ERROR;
}
- if (remote_msg.msg_len > HELLO_MSG_LEN_MIN) {
+ if (remote_msg.msg_len > HELLO_V2_MSG_LEN_MIN) {
// Drain unknown trailing bytes so they don't pollute subsequent
// reads (e.g. the upcoming ACK message). v2 base fields already
// carry enough information for negotiation; unknown trailing
// bytes are treated as optional hints that v2 safely ignores.
- size_t ext_len = remote_msg.msg_len - HELLO_MSG_LEN_MIN;
+ size_t ext_len = remote_msg.msg_len - HELLO_V2_MSG_LEN_MIN;
if (DrainBytes(ep, ext_len) < 0) {
- return -1;
+ return RemoteHelloResult::ERROR;
}
}
if (!ValidHelloMessage(remote_msg)) {
- *negotiated = false;
- return 0;
+ return RemoteHelloResult::FALLBACK;
}
- *negotiated = true;
TranslateV2Hello(remote_msg, remote);
- return 0;
+ return RemoteHelloResult::NEGOTIATED;
}
int DrainBytes(RdmaEndpoint* ep, size_t n) {
@@ -153,12 +142,12 @@ int DrainBytes(RdmaEndpoint* ep, size_t n) {
int RdmaHandshakeClientV2::SendLocalHello() {
RdmaEndpoint* ep = _ep;
- uint8_t data[RDMA_HELLO_V2_MSG_LEN];
+ uint8_t data[HELLO_V2_MSG_LEN_MIN];
v2_wire::HelloMessage local_msg{};
- local_msg.msg_len = RDMA_HELLO_V2_MSG_LEN;
- local_msg.hello_ver = RDMA_HELLO_V2_VERSION;
- local_msg.impl_ver = RDMA_IMPL_V2_VERSION;
+ local_msg.msg_len = HELLO_V2_MSG_LEN_MIN;
+ local_msg.hello_ver = HELLO_V2_VERSION;
+ local_msg.impl_ver = IMPL_V2_VERSION;
local_msg.block_size = g_rdma_recv_block_size;
local_msg.sq_size = ep->_sq_size;
local_msg.rq_size = ep->_rq_size;
@@ -170,36 +159,70 @@ int RdmaHandshakeClientV2::SendLocalHello() {
// Only happens in UT
local_msg.qp_num = 0;
}
- fast_memcpy(data, MAGIC_STR, 4);
+ fast_memcpy(data, HELLO_MAGIC, 4);
local_msg.Serialize((char*)data + 4);
- return ep->WriteToFd(data, RDMA_HELLO_V2_MSG_LEN);
+ return ep->WriteToFd(data, HELLO_V2_MSG_LEN_MIN);
}
-int RdmaHandshakeClientV2::ReceiveAndParseRemoteHello(ParsedHello* remote,
- bool* negotiated) {
- RdmaEndpoint* ep = _ep;
-
- // Read and verify magic (the endpoint did NOT pre-read magic on the
client side).
- uint8_t magic[MAGIC_STR_LEN];
- if (ep->ReadFromFd(magic, MAGIC_STR_LEN) < 0) {
- return -1;
+RemoteHelloResult
RdmaHandshakeClientV2::ReceiveAndParseRemoteHello(ParsedHello* remote) {
+ uint8_t magic[HELLO_MAGIC_LEN];
+ if (_ep->ReadFromFd(magic, HELLO_MAGIC_LEN) < 0) {
+ return RemoteHelloResult::ERROR;
}
- if (memcmp(magic, MAGIC_STR, MAGIC_STR_LEN) != 0) {
+ if (memcmp(magic, HELLO_MAGIC, HELLO_MAGIC_LEN) != 0) {
errno = EPROTO;
- return -1;
+ return RemoteHelloResult::ERROR;
}
- return v2_wire::ReadBodyAndNegotiate(ep, remote, negotiated);
+
+ return v2_wire::ReadBodyAndNegotiate(_ep, remote);
}
-int RdmaHandshakeServerV2::ReceiveAndParseRemoteHello(ParsedHello* remote,
bool* negotiated) {
- // Magic already consumed by ProcessHandshakeAtServer.
- return v2_wire::ReadBodyAndNegotiate(_ep, remote, negotiated);
+// Parse one complete v2 client hello out of `_source` (non-blocking).
+// v2 hello: [ "RDMA" 4B ][ msg_len 2B ][ 34B ... ], base total = 40B.
+RemoteHelloResult
RdmaHandshakeServerV2::ReceiveAndParseRemoteHello(ParsedHello* remote) {
+ butil::IOBuf* source = _source;
+ constexpr size_t HDR_LEN = HELLO_MAGIC_LEN + 2;
+ if (source->size() < HDR_LEN) {
+ // msg_len has not fully arrived yet.
+ return RemoteHelloResult::NEED_MORE;
+ }
+
+ uint8_t hdr[HDR_LEN];
+ CHECK_EQ(source->copy_to(hdr, sizeof(hdr)), sizeof(hdr));
+
+ uint16_t msg_len = 0;
+ butil::RawUnpacker(hdr + HELLO_MAGIC_LEN).unpack16(msg_len);
+ if (msg_len < HELLO_V2_MSG_LEN_MIN || msg_len > HELLO_V2_MSG_LEN_MAX) {
+ errno = EPROTO;
+ return RemoteHelloResult::ERROR;
+ }
+ if (source->size() < msg_len) {
+ // Full message has not fully arrived yet.
+ return RemoteHelloResult::NEED_MORE;
+ }
+
+ // Consume the whole hello: magic + 36B base body + optional extension.
+ CHECK_EQ(source->pop_front(HELLO_MAGIC_LEN), HELLO_MAGIC_LEN);
+ uint8_t body[HELLO_V2_MSG_LEN_MIN - HELLO_MAGIC_LEN]; // 36B
+ CHECK_EQ(source->cutn(body, sizeof(body)), sizeof(body));
+ if (!source->empty()) {
+ // Drain unknown trailing bytes.
+ source->clear();
+ }
+
+ v2_wire::HelloMessage remote_msg{};
+ remote_msg.Deserialize(body);
+ if (!v2_wire::ValidHelloMessage(remote_msg)) {
+ return RemoteHelloResult::FALLBACK;
+ }
+ v2_wire::TranslateV2Hello(remote_msg, remote);
+ return RemoteHelloResult::NEGOTIATED;
}
int RdmaHandshakeServerV2::SendLocalHello() {
- uint8_t data[RDMA_HELLO_V2_MSG_LEN];
+ uint8_t data[HELLO_V2_MSG_LEN_MIN];
v2_wire::HelloMessage local_msg{};
- local_msg.msg_len = RDMA_HELLO_V2_MSG_LEN;
+ local_msg.msg_len = HELLO_V2_MSG_LEN_MIN;
auto rdma_transport =
static_cast<RdmaTransport*>(_ep->_socket->_transport.get());
if (rdma_transport->_rdma_state == RdmaTransport::RDMA_OFF) {
local_msg.hello_ver = 0;
@@ -211,8 +234,8 @@ int RdmaHandshakeServerV2::SendLocalHello() {
memset(local_msg.gid.raw, 0, sizeof(local_msg.gid.raw));
local_msg.qp_num = 0;
} else {
- local_msg.hello_ver = RDMA_HELLO_V2_VERSION;
- local_msg.impl_ver = RDMA_IMPL_V2_VERSION;
+ local_msg.hello_ver = HELLO_V2_VERSION;
+ local_msg.impl_ver = IMPL_V2_VERSION;
local_msg.block_size = g_rdma_recv_block_size;
local_msg.sq_size = _ep->_sq_size;
local_msg.rq_size = _ep->_rq_size;
@@ -225,9 +248,9 @@ int RdmaHandshakeServerV2::SendLocalHello() {
local_msg.qp_num = 0;
}
}
- fast_memcpy(data, MAGIC_STR, 4);
+ fast_memcpy(data, HELLO_MAGIC, 4);
local_msg.Serialize((char*)data + 4);
- return _ep->WriteToFd(data, RDMA_HELLO_V2_MSG_LEN);
+ return _ep->WriteToFd(data, HELLO_V2_MSG_LEN_MIN);
}
namespace v3_wire {
@@ -263,8 +286,7 @@ void FillLocalRdmaHello(const RdmaEndpoint* ep, RdmaHello*
msg) {
msg->set_rq_size(ep->_rq_size);
msg->set_lid(GetRdmaLid());
ibv_gid gid = GetRdmaGid();
- msg->set_gid(std::string(reinterpret_cast<const char*>(gid.raw),
- sizeof(gid.raw)));
+ msg->set_gid(reinterpret_cast<const char*>(gid.raw), sizeof(gid.raw));
if (BAIDU_LIKELY(ep->_resource)) {
msg->set_qp_num(ep->_resource->qp->qp_num);
} else {
@@ -274,13 +296,13 @@ void FillLocalRdmaHello(const RdmaEndpoint* ep,
RdmaHello* msg) {
}
int ReadAndParseV3Hello(RdmaEndpoint* ep, RdmaHello* out) {
- uint8_t size_buf[RDMA_HELLO_V3_PB_SIZE_LEN];
- if (ep->ReadFromFd(size_buf, RDMA_HELLO_V3_PB_SIZE_LEN) < 0) {
+ uint8_t size_buf[HELLO_V3_PB_SIZE_LEN];
+ if (ep->ReadFromFd(size_buf, HELLO_V3_PB_SIZE_LEN) < 0) {
return -1;
}
uint32_t pb_size = butil::NetToHost32(
*reinterpret_cast<const uint32_t*>(size_buf));
- if (pb_size == 0 || pb_size > RDMA_HELLO_V3_MAX_PB_SIZE) {
+ if (pb_size == 0 || pb_size > HELLO_V3_MAX_PB_SIZE) {
errno = EPROTO;
return -1;
}
@@ -300,16 +322,16 @@ int ReadAndParseV3Hello(RdmaEndpoint* ep, RdmaHello* out)
{
int WriteV3Hello(RdmaEndpoint* ep, const RdmaHello& msg) {
uint32_t pb_size = static_cast<uint32_t>(msg.ByteSizeLong());
- if (pb_size > RDMA_HELLO_V3_MAX_PB_SIZE) {
+ if (pb_size > HELLO_V3_MAX_PB_SIZE) {
errno = EPROTO;
return -1;
}
// [ "RDM3" 4B ][ pb_size 4B (big-endian) ][ RdmaHello protobuf bytes ]
butil::IOBuf packet;
- packet.append(MAGIC_STR_V3, MAGIC_STR_LEN);
+ packet.append(HELLO_MAGIC_V3, HELLO_MAGIC_LEN);
uint32_t pb_size_be = butil::HostToNet32(pb_size);
- packet.append(&pb_size_be, RDMA_HELLO_V3_PB_SIZE_LEN);
+ packet.append(&pb_size_be, HELLO_V3_PB_SIZE_LEN);
butil::IOBufAsZeroCopyOutputStream output(&packet);
if (!msg.SerializeToZeroCopyStream(&output)) {
LOG(ERROR) << "Failed to serialize RdmaHello";
@@ -336,48 +358,83 @@ int RdmaHandshakeClientV3::SendLocalHello() {
return v3_wire::WriteV3Hello(_ep, local_msg);
}
-int RdmaHandshakeClientV3::ReceiveAndParseRemoteHello(ParsedHello* remote,
- bool* negotiated) {
- uint8_t magic[MAGIC_STR_LEN];
- if (_ep->ReadFromFd(magic, MAGIC_STR_LEN) < 0) {
- return -1;
+RemoteHelloResult
RdmaHandshakeClientV3::ReceiveAndParseRemoteHello(ParsedHello* remote) {
+ uint8_t magic[HELLO_MAGIC_LEN];
+ if (_ep->ReadFromFd(magic, HELLO_MAGIC_LEN) < 0) {
+ return RemoteHelloResult::ERROR;
}
- if (memcmp(magic, MAGIC_STR_V3, MAGIC_STR_LEN) != 0) {
+ if (memcmp(magic, HELLO_MAGIC_V3, HELLO_MAGIC_LEN) != 0) {
errno = EPROTO;
- return -1;
+ return RemoteHelloResult::ERROR;
}
RdmaHello remote_msg{};
if (v3_wire::ReadAndParseV3Hello(_ep, &remote_msg) < 0) {
- return -1;
+ return RemoteHelloResult::ERROR;
}
if (!v3_wire::ValidRdmaHello(remote_msg)) {
- *negotiated = false;
- return 0;
+ return RemoteHelloResult::FALLBACK;
}
- *negotiated = true;
v3_wire::TranslateHello(remote_msg, remote);
- return 0;
+ return RemoteHelloResult::NEGOTIATED;
}
-int RdmaHandshakeServerV3::ReceiveAndParseRemoteHello(ParsedHello* remote,
bool* negotiated) {
- // Magic already consumed by ProcessHandshakeAtServer.
- RdmaHello remote_msg{};
- if (v3_wire::ReadAndParseV3Hello(_ep, &remote_msg) < 0) {
- return -1;
+// Parse one complete v3 client hello out of `_source` (non-blocking).
+// v3 hello: [ "RDM3" 4B ][ pb_size 4B (big-endian) ][ RdmaHello ]
+RemoteHelloResult
RdmaHandshakeServerV3::ReceiveAndParseRemoteHello(ParsedHello* remote) {
+ constexpr size_t HDR_LEN = HELLO_MAGIC_LEN + HELLO_V3_PB_SIZE_LEN;
+ if (_source->size() < HDR_LEN) {
+ // pb_size has not fully arrived yet.
+ return RemoteHelloResult::NEED_MORE;
+ }
+
+ uint8_t hdr[HDR_LEN];
+ CHECK_EQ(_source->copy_to(hdr, sizeof(hdr)), sizeof(hdr));
+
+ uint32_t pb_size = butil::NetToHost32(
+ *reinterpret_cast<const uint32_t*>(hdr + HELLO_MAGIC_LEN));
+ if (pb_size == 0 || pb_size > HELLO_V3_MAX_PB_SIZE) {
+ errno = EPROTO;
+ return RemoteHelloResult::ERROR;
+ }
+ size_t total = HDR_LEN + pb_size;
+ if (_source->size() < total) {
+ // Full message has not fully arrived yet.
+ return RemoteHelloResult::NEED_MORE;
+ }
+
+ CHECK_EQ(_source->cutn(hdr, HDR_LEN), HDR_LEN);
+ butil::IOBuf pb;
+ CHECK_EQ(_source->cutn(&pb, pb_size), pb_size);
+ RdmaHello remote_msg;
+ butil::IOBufAsZeroCopyInputStream input(pb);
+ if (!remote_msg.ParseFromZeroCopyStream(&input)) {
+ LOG(ERROR) << "Failed to parse RdmaHello";
+ errno = EPROTO;
+ return RemoteHelloResult::ERROR;
}
if (!v3_wire::ValidRdmaHello(remote_msg)) {
- *negotiated = false;
- return 0;
+ return RemoteHelloResult::FALLBACK;
}
- *negotiated = true;
v3_wire::TranslateHello(remote_msg, remote);
- return 0;
+ return RemoteHelloResult::NEGOTIATED;
}
int RdmaHandshakeServerV3::SendLocalHello() {
RdmaHello local_msg{};
- v3_wire::FillLocalRdmaHello(_ep, &local_msg);
+ auto rdma_transport =
static_cast<RdmaTransport*>(_ep->_socket->_transport.get());
+ if (rdma_transport->_rdma_state == RdmaTransport::RDMA_OFF) {
+ // Un-negotiable hello: all body fields are zero so the client's
+ // rejects it and downgrades to TCP on the same connection.
+ local_msg.set_block_size(0);
+ local_msg.set_sq_size(0);
+ local_msg.set_rq_size(0);
+ local_msg.set_lid(0);
+ local_msg.set_gid(std::string(sizeof(ibv_gid), '\0'));
+ local_msg.set_qp_num(0);
+ } else {
+ v3_wire::FillLocalRdmaHello(_ep, &local_msg);
+ }
return v3_wire::WriteV3Hello(_ep, local_msg);
}
@@ -392,14 +449,16 @@ std::unique_ptr<RdmaHandshake>
CreateClientHandshake(RdmaEndpoint* ep) {
}
std::unique_ptr<RdmaHandshake> CreateServerHandshakeByMagic(
- RdmaEndpoint* ep, const uint8_t magic[MAGIC_STR_LEN]) {
- if (memcmp(magic, MAGIC_STR, MAGIC_STR_LEN) == 0) {
- return std::unique_ptr<RdmaHandshake>(new RdmaHandshakeServerV2(ep));
+ RdmaEndpoint* ep, butil::IOBuf* source, const uint8_t
magic[HELLO_MAGIC_LEN]) {
+ if (memcmp(magic, HELLO_MAGIC, HELLO_MAGIC_LEN) == 0) {
+ return std::unique_ptr<RdmaHandshake>(
+ new RdmaHandshakeServerV2(ep, source));
}
- if (memcmp(magic, MAGIC_STR_V3, MAGIC_STR_LEN) == 0) {
- return std::unique_ptr<RdmaHandshake>(new RdmaHandshakeServerV3(ep));
+ if (memcmp(magic, HELLO_MAGIC_V3, HELLO_MAGIC_LEN) == 0) {
+ return std::unique_ptr<RdmaHandshake>(
+ new RdmaHandshakeServerV3(ep, source));
}
- return nullptr;
+ return NULL;
}
} // namespace rdma
diff --git a/src/brpc/rdma/rdma_handshake.h b/src/brpc/rdma/rdma_handshake.h
index 5f36a9e6..b35a2378 100644
--- a/src/brpc/rdma/rdma_handshake.h
+++ b/src/brpc/rdma/rdma_handshake.h
@@ -20,20 +20,20 @@
#if BRPC_WITH_RDMA
-#include <cstddef>
-#include <cstdint>
#include <memory>
#include <infiniband/verbs.h>
#include "butil/macros.h"
+#include "brpc/rdma/rdma_handshake_constants.h"
+
+namespace butil {
+class IOBuf;
+}
namespace brpc {
namespace rdma {
class RdmaEndpoint;
-// Length of the RDMA handshake magic string (e.g. "RDMA", "RDM3").
-static const size_t MAGIC_STR_LEN = 4;
-
// Wire-format-agnostic representation of a peer's hello message.
// Each protocol version (v2 binary, v3 protobuf) translates its own
// wire format into this struct so the state-machine driver in
@@ -48,18 +48,19 @@ struct ParsedHello {
uint32_t qp_num;
};
-namespace v2_wire {
+// Result of reading/parsing a peer's hello (see ReceiveAndParseRemoteHello).
+enum class RemoteHelloResult {
+ // A full hello was read and negotiation succeeded.
+ NEGOTIATED,
+ // A full hello was read but negotiation failed.
+ FALLBACK,
+ // (server only) Not enough data yet.
+ NEED_MORE,
+ // IO/protocol error (errno set).
+ ERROR,
+};
-// Wire constants for the v2 hello.
-//
-// HELLO_MSG_LEN_MIN: total length of the base v2 hello (4B magic +
-// 36B HelloMessage). Anything shorter than this is malformed.
-// HELLO_MSG_LEN_MAX: upper bound for the entire v2 hello message
-// length declared by HelloMessage::msg_len. Anything beyond this is
-// treated as a protocol error and the connection is closed without
-// attempting to drain.
-static constexpr size_t HELLO_MSG_LEN_MIN = 40;
-static constexpr size_t HELLO_MSG_LEN_MAX = 4096;
+namespace v2_wire {
// v2 binary HelloMessage.
struct HelloMessage {
@@ -79,19 +80,16 @@ struct HelloMessage {
} // namespace v2_wire
-// Abstract base class of an RDMA handshake.
-//
-// Acts as the protocol-version dispatch point for the state machine
-// driven by RdmaEndpoint::ProcessHandshakeAt{Client,Server}.
+// Base class of an RDMA handshake, shared by both roles.
class RdmaHandshake {
public:
- explicit RdmaHandshake(RdmaEndpoint* ep) : _ep(ep) {}
+ RdmaHandshake(RdmaEndpoint* ep, int version) : _ep(ep), _version(version)
{}
virtual ~RdmaHandshake() = default;
DISALLOW_COPY_AND_ASSIGN(RdmaHandshake);
// Wire-level protocol version (2 for "RDMA", 3 for "RDM3").
- virtual int ProtocolVersion() const = 0;
+ int ProtocolVersion() const { return _version; }
// Build and send the local hello (including the protocol magic).
// Returns 0 on success, -1 on IO error (errno set).
@@ -104,68 +102,56 @@ public:
// - v3: qp_num==0 so the peer's ValidRdmaHello rejects it.
virtual int SendLocalHello() = 0;
- // Read the peer's hello, validate it, and translate into ParsedHello.
- //
- // Role-specific semantics:
- // - Client subclasses: read & verify the 4B magic first, then the
- // body. (The endpoint did NOT pre-read the magic on the client
- // side.)
- // - Server subclasses: read ONLY the body. The 4B magic was
- // already consumed by ProcessHandshakeAtServer and was used to
- // pick `this` from CreateServerHandshakeByMagic; re-reading
- // would deadlock.
- //
- // Outputs:
- // *negotiated -- true if the remote hello is structurally valid
- // AND passes per-protocol negotiation checks;
- // false means the peer asked for fallback or sent
- // something we can't honor.
- // Returns:
- // 0 -- IO/parsing layer OK; check *negotiated and *remote.
- // -1 -- IO error or unrecoverable protocol error (errno set).
- virtual int ReceiveAndParseRemoteHello(ParsedHello* remote, bool*
negotiated) = 0;
+ // Read and parse the peer's hello into *remote.
+ virtual RemoteHelloResult ReceiveAndParseRemoteHello(ParsedHello* remote)
= 0;
protected:
RdmaEndpoint* _ep;
+ int _version;
+};
+
+// Server-side handshake base: parses the remote hello non-blockingly out of
+// `_source` (an IOBuf filled by InputMessenger), never touching the fd.
+class ServerRdmaHandshake : public RdmaHandshake {
+public:
+ ServerRdmaHandshake(RdmaEndpoint* ep, butil::IOBuf* source, int version)
+ : RdmaHandshake(ep, version), _source(source) {}
+
+protected:
+ butil::IOBuf* _source;
};
// v2 handshake (legacy "RDMA" magic, 36B binary HelloMessage).
class RdmaHandshakeClientV2 : public RdmaHandshake {
public:
- using RdmaHandshake::RdmaHandshake;
- int ProtocolVersion() const override { return 2; }
-
+ explicit RdmaHandshakeClientV2(RdmaEndpoint* ep) : RdmaHandshake(ep, 2) {}
int SendLocalHello() override;
- int ReceiveAndParseRemoteHello(ParsedHello* remote, bool* negotiated)
override;
+ RemoteHelloResult ReceiveAndParseRemoteHello(ParsedHello* remote) override;
};
-class RdmaHandshakeServerV2 : public RdmaHandshake {
+class RdmaHandshakeServerV2 : public ServerRdmaHandshake {
public:
- using RdmaHandshake::RdmaHandshake;
- int ProtocolVersion() const override { return 2; }
-
+ RdmaHandshakeServerV2(RdmaEndpoint* ep, butil::IOBuf* source)
+ : ServerRdmaHandshake(ep, source, 2) {}
int SendLocalHello() override;
- int ReceiveAndParseRemoteHello(ParsedHello* remote, bool* negotiated)
override;
+ RemoteHelloResult ReceiveAndParseRemoteHello(ParsedHello* remote) override;
};
// v3 handshake (new "RDM3" magic, protobuf RdmaHello).
// [ "RDM3" 4B ][ pb_size 4B (big-endian) ][ RdmaHello protobuf bytes ]
class RdmaHandshakeClientV3 : public RdmaHandshake {
public:
- using RdmaHandshake::RdmaHandshake;
- int ProtocolVersion() const override { return 3; }
-
+ explicit RdmaHandshakeClientV3(RdmaEndpoint* ep) : RdmaHandshake(ep, 3) {}
int SendLocalHello() override;
- int ReceiveAndParseRemoteHello(ParsedHello* remote, bool* negotiated)
override;
+ RemoteHelloResult ReceiveAndParseRemoteHello(ParsedHello* remote) override;
};
-class RdmaHandshakeServerV3 : public RdmaHandshake {
+class RdmaHandshakeServerV3 : public ServerRdmaHandshake {
public:
- using RdmaHandshake::RdmaHandshake;
- int ProtocolVersion() const override { return 3; }
-
+ RdmaHandshakeServerV3(RdmaEndpoint* ep, butil::IOBuf* source)
+ : ServerRdmaHandshake(ep, source, 3) {}
int SendLocalHello() override;
- int ReceiveAndParseRemoteHello(ParsedHello* remote, bool* negotiated)
override;
+ RemoteHelloResult ReceiveAndParseRemoteHello(ParsedHello* remote) override;
};
// Factory methods
@@ -183,7 +169,7 @@ std::unique_ptr<RdmaHandshake>
CreateClientHandshake(RdmaEndpoint* ep);
// "RDMA" -> RdmaHandshakeServerV2
// "RDM3" -> RdmaHandshakeServerV3
std::unique_ptr<RdmaHandshake> CreateServerHandshakeByMagic(
- RdmaEndpoint* ep, const uint8_t magic[MAGIC_STR_LEN]);
+ RdmaEndpoint* ep, butil::IOBuf* source, const uint8_t
magic[HELLO_MAGIC_LEN]);
} // namespace rdma
} // namespace brpc
diff --git a/src/brpc/rdma/rdma_handshake_constants.h
b/src/brpc/rdma/rdma_handshake_constants.h
new file mode 100644
index 00000000..aa9811b9
--- /dev/null
+++ b/src/brpc/rdma/rdma_handshake_constants.h
@@ -0,0 +1,56 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#ifndef BRPC_RDMA_RDMA_HANDSHAKE_CONSTANTS_H
+#define BRPC_RDMA_RDMA_HANDSHAKE_CONSTANTS_H
+
+namespace brpc {
+namespace rdma {
+
+// Handshake magic strings and their (shared) length.
+// v2 ("RDMA"): [ "RDMA" 4B ][ msg_len 2B ][ 34B body ], total 40B.
+// v3 ("RDM3"): [ "RDM3" 4B ][ pb_size 4B (big-endian) ][ RdmaHello bytes ].
+constexpr const char* HELLO_MAGIC = "RDMA";
+constexpr const char* HELLO_MAGIC_V3 = "RDM3";
+constexpr size_t HELLO_MAGIC_LEN = 4;
+
+// v2 hello version fields. A valid v2 hello must carry exactly these.
+constexpr uint16_t HELLO_V2_VERSION = 2;
+constexpr uint16_t IMPL_V2_VERSION = 1;
+
+// v2 hello length bounds. HELLO_V2_MSG_LEN_MIN is the total length of the
+// base v2 hello (4B magic + 36B HelloMessage); anything shorter is malformed.
+// HELLO_V2_MSG_LEN_MAX caps the msg_len declared by the peer; anything
+// beyond it is treated as a protocol error and the connection is closed
without
+// attempting to drain.
+constexpr size_t HELLO_V2_MSG_LEN_MIN = 40;
+constexpr size_t HELLO_V2_MSG_LEN_MAX = 4096;
+
+// v3 hello framing: a 4B big-endian protobuf size prefix, capped to avoid
+// waiting forever for bytes that never come.
+constexpr size_t HELLO_V3_PB_SIZE_LEN = 4;
+constexpr size_t HELLO_V3_MAX_PB_SIZE = 8192;
+
+// Handshake ACK (shared by all versions): a single 4B big-endian flags word;
+// bit 0 (HELLO_ACK_RDMA_OK) means the sender wants to use RDMA.
+constexpr size_t HELLO_ACK_LEN = 4;
+constexpr uint32_t HELLO_ACK_RDMA_OK = 0x1;
+
+} // namespace rdma
+} // namespace brpc
+
+#endif // BRPC_RDMA_RDMA_HANDSHAKE_CONSTANTS_H
diff --git a/src/brpc/rdma/rdma_handshake_server.cpp
b/src/brpc/rdma/rdma_handshake_server.cpp
new file mode 100644
index 00000000..4072a11c
--- /dev/null
+++ b/src/brpc/rdma/rdma_handshake_server.cpp
@@ -0,0 +1,214 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "brpc/rdma/rdma_handshake_server.h"
+
+#include <limits>
+#include <string>
+#include <string.h>
+#include "butil/iobuf.h"
+#include "butil/logging.h"
+#include "butil/object_pool.h"
+#include "butil/raw_pack.h"
+#include "butil/sys_byteorder.h"
+#include "brpc/socket.h"
+#include "brpc/rdma/rdma_handshake.pb.h"
+#include "brpc/rdma/rdma_handshake_constants.h"
+#if BRPC_WITH_RDMA
+#include "brpc/rdma/rdma_endpoint.h"
+#endif
+
+namespace brpc {
+namespace rdma {
+
+ServerHandshakeContext* ServerHandshakeContext::Create() {
+ return butil::get_object<ServerHandshakeContext>();
+}
+
+void ServerHandshakeContext::Destroy() {
+ butil::return_object(this);
+}
+
+// Fallback-only server handshake. Used for any connection that is NOT in RDMA
+// mode: builds without RDMA (no RdmaEndpoint exists at all), and RDMA-enabled
+// builds where this particular connection is plain TCP. Every RDMA client that
+// reaches here is answered with an un-negotiable hello and asked to downgrade
+// to TCP on the same connection, then its ACK is drained.
+
+// An intentionally-invalid v2 hello_ver: the client's ValidHelloMessage()
+// requires hello_ver==2, so it rejects this and falls back.
+static constexpr uint16_t V2_HELLO_VERSION_INVALID =
std::numeric_limits<uint16_t>::max();
+// Length of the gid field (== sizeof(ibv_gid)), spelled as a literal so this
+// compile-switch-independent file stays free of <infiniband/verbs.h>.
+static constexpr size_t V3_GID_LEN = 16;
+
+// Consume one complete v2 client hello ("RDMA" + 2B msg_len + body) from
+// `source` without interpreting its content (fallback does not negotiate).
+// Returns 1 (consumed), 0 (not enough data yet, nothing consumed) or -1
(error).
+static int DrainClientHelloV2(butil::IOBuf* source) {
+ constexpr size_t HDR_LEN = HELLO_MAGIC_LEN + 2;
+ if (source->size() < HDR_LEN) {
+ return 0;
+ }
+
+ uint8_t hdr[HDR_LEN];
+ CHECK_EQ(source->copy_to(hdr, sizeof(hdr)), sizeof(hdr));
+
+ uint16_t msg_len = 0;
+ butil::RawUnpacker(hdr + HELLO_MAGIC_LEN).unpack16(msg_len);
+ if (msg_len < HELLO_V2_MSG_LEN_MIN || msg_len > HELLO_V2_MSG_LEN_MAX) {
+ return -1;
+ }
+
+ if (source->size() < msg_len) {
+ return 0;
+ }
+
+ CHECK_EQ(source->pop_front(msg_len), msg_len);
+ return 1;
+}
+
+// Consume one complete v3 client hello ("RDM3" + 4B pb_size + protobuf body)
+// from `source` without interpreting its content.
+// Returns 1 (consumed), 0 (not enough data yet, nothing consumed) or -1
(error).
+static int DrainClientHelloV3(butil::IOBuf* source) {
+ constexpr size_t HDR_LEN = HELLO_MAGIC_LEN + HELLO_V3_PB_SIZE_LEN;
+ if (source->size()< HDR_LEN) {
+ return 0;
+ }
+
+ uint8_t hdr[HDR_LEN];
+ CHECK_EQ(source->copy_to(hdr, sizeof(hdr)), sizeof(hdr));
+
+ uint32_t pb_size = butil::NetToHost32(
+ *reinterpret_cast<const uint32_t*>(hdr + HELLO_MAGIC_LEN));
+ if (pb_size == 0 || pb_size > HELLO_V3_MAX_PB_SIZE) {
+ return -1;
+ }
+
+ const size_t total = HDR_LEN + pb_size;
+ if (source->size() < total) {
+ return 0;
+ }
+
+ CHECK_EQ(source->pop_front(total), total);
+ return 1;
+}
+
+// Reply an un-negotiable hello so the client downgrades to TCP. All body
fields
+// are zero/invalid so the client's validity check rejects it.
+// Returns 0 on success, -1 otherwise.
+static int SendUnnegotiableHello(Socket* socket, int version) {
+ butil::IOBuf packet;
+ if (version == 2) {
+ // magic "RDMA" + msg_len(=40) + invalid hello_ver; rest stays zero.
+ // NOTE: msg_len is the length of the WHOLE hello INCLUDING the 4B
magic
+ // (== HELLO_V2_MSG_LEN_MIN), not just the body; the client rejects any
+ // msg_len < HELLO_V2_MSG_LEN_MIN as a protocol error.
+ packet.append(HELLO_MAGIC, HELLO_MAGIC_LEN);
+ char reply[HELLO_V2_MSG_LEN_MIN - HELLO_MAGIC_LEN]{};
+ butil::RawPacker(reply).pack16(HELLO_V2_MSG_LEN_MIN)
+ .pack16(V2_HELLO_VERSION_INVALID);
+ packet.append(reply, sizeof(reply));
+ } else {
+ // "RDM3" + pb_size + RdmaHello with block_size==0 & qp_num==0 so the
+ // client's ValidRdmaHello() returns false.
+ RdmaHello reply_msg;
+ reply_msg.set_block_size(0);
+ reply_msg.set_sq_size(0);
+ reply_msg.set_rq_size(0);
+ reply_msg.set_lid(0);
+ reply_msg.set_gid(std::string(V3_GID_LEN, '\0'));
+ reply_msg.set_qp_num(0);
+ packet.append(HELLO_MAGIC_V3, HELLO_MAGIC_LEN);
+ uint32_t pb_size_be =
+
butil::HostToNet32(static_cast<uint32_t>(reply_msg.ByteSizeLong()));
+ packet.append(&pb_size_be, sizeof(pb_size_be));
+ butil::IOBufAsZeroCopyOutputStream output(&packet);
+ if (!reply_msg.SerializeToZeroCopyStream(&output)) {
+ LOG(WARNING) << "Fail to serialize RDMA v3 fallback hello";
+ return -1;
+ }
+ }
+
+ if (socket->Write(&packet) != 0) {
+ PLOG(WARNING) << "Fail to send RDMA fallback hello to " <<
socket->description();
+ return -1;
+ }
+ return 0;
+}
+
+// Fallback handshake for connections that are NOT in RDMA mode.
+static ParseResult FallbackServerHandshake(butil::IOBuf* source, Socket*
socket) {
+ if (socket->parsing_context() == NULL) {
+ if (source->size() < HELLO_MAGIC_LEN) {
+ return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
+ }
+ // Phase 1: consume the client hello and reply an un-negotiable hello.
+ uint8_t magic[HELLO_MAGIC_LEN];
+ CHECK_EQ(source->copy_to(magic, HELLO_MAGIC_LEN), HELLO_MAGIC_LEN);
+
+ int version;
+ if (memcmp(magic, HELLO_MAGIC, HELLO_MAGIC_LEN) == 0) {
+ version = 2;
+ } else if (memcmp(magic, HELLO_MAGIC_V3, HELLO_MAGIC_LEN) == 0) {
+ version = 3;
+ } else {
+ return MakeParseError(PARSE_ERROR_TRY_OTHERS);
+ }
+
+ const int r = version == 2 ? DrainClientHelloV2(source) :
DrainClientHelloV3(source);
+ if (r == 0) {
+ // Hello not complete yet; keep the buffer intact and retry later.
+ return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
+ }
+ if (r < 0) {
+ return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
+ }
+ if (SendUnnegotiableHello(socket, version) < 0) {
+ return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
+ }
+ // Wait for the client ACK across subsequent reads.
+ socket->reset_parsing_context(ServerHandshakeContext::Create());
+ return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
+ }
+
+ // Phase 2: drain the 4B ACK.
+ if (source->size() < HELLO_ACK_LEN) {
+ return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
+ }
+ CHECK_EQ(source->pop_front(HELLO_ACK_LEN), HELLO_ACK_LEN);
+ // Handshake done (downgraded to TCP); drop the context and let
+ // InputMessenger parse the following real RPC.
+ socket->reset_parsing_context(NULL);
+ return MakeParseError(PARSE_ERROR_TRY_OTHERS);
+}
+
+ParseResult ExecuteServerHandshake(butil::IOBuf* source, Socket* socket) {
+ // Only RDMA-mode connections carry a live RdmaEndpoint and run the real
+ // handshake. A connection that is not in RDMA mode (RDMA compiled in but
+ // this connection is plain TCP, or RDMA not compiled at all) falls back.
+#if BRPC_WITH_RDMA
+ if (socket->socket_mode() == SOCKET_MODE_RDMA) {
+ return RdmaEndpoint::ExecuteServerHandshake(source, socket);
+ }
+#endif
+ return FallbackServerHandshake(source, socket);
+}
+
+} // namespace rdma
+} // namespace brpc
diff --git a/src/brpc/rdma/rdma_handshake_server.h
b/src/brpc/rdma/rdma_handshake_server.h
new file mode 100644
index 00000000..705aa893
--- /dev/null
+++ b/src/brpc/rdma/rdma_handshake_server.h
@@ -0,0 +1,48 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#ifndef BRPC_RDMA_RDMA_HANDSHAKE_SERVER_H
+#define BRPC_RDMA_RDMA_HANDSHAKE_SERVER_H
+
+#include "brpc/destroyable.h"
+#include "brpc/parse_result.h"
+
+namespace butil {
+class IOBuf;
+}
+
+namespace brpc {
+class Socket;
+namespace rdma {
+
+// State kept across multiple parse calls of the server handshake.
+struct ServerHandshakeContext : public Destroyable {
+ static ServerHandshakeContext* Create();
+ void Destroy() override;
+};
+
+// The single server-side RDMA handshake entry for policy::ParseRdmaHandshake.
+// Returns a ParseResult ready to be handed back from the protocol parser:
+// - not an RDMA magic / handshake finished -> PARSE_ERROR_TRY_OTHERS;
+// - an RDMA magic but not enough bytes yet -> PARSE_ERROR_NOT_ENOUGH_DATA;
+// - IO/protocol error -> PARSE_ERROR_ABSOLUTELY_WRONG.
+ParseResult ExecuteServerHandshake(butil::IOBuf* source, Socket* socket);
+
+} // namespace rdma
+} // namespace brpc
+
+#endif // BRPC_RDMA_RDMA_HANDSHAKE_SERVER_H
diff --git a/src/brpc/rdma_transport.cpp b/src/brpc/rdma_transport.cpp
index 88d89a7b..b172e47f 100644
--- a/src/brpc/rdma_transport.cpp
+++ b/src/brpc/rdma_transport.cpp
@@ -20,6 +20,7 @@
#include "brpc/rdma_transport.h"
#include "brpc/event_dispatcher.h"
#include "brpc/tcp_transport.h"
+#include "brpc/input_messenger.h"
#include "brpc/rdma/rdma_endpoint.h"
#include "brpc/rdma/rdma_helper.h"
@@ -48,7 +49,16 @@ void RdmaTransport::Init(Socket *socket, const SocketOptions
&options) {
_default_connect = options.app_connect;
_on_edge_trigger = options.on_edge_triggered_events;
if (options.need_on_edge_trigger && _on_edge_trigger == NULL) {
- _on_edge_trigger = rdma::RdmaEndpoint::OnNewDataFromTcp;
+ // Server-side RDMA sockets drive the handshake through the standard
+ // InputMessenger path (ParseRdmaHandshake), so they use OnNewMessages
+ // just like TCP sockets. Only client-side sockets, whose handshake
+ // (ProcessHandshakeAtClient) is an active blocking bthread relying on
+ // _read_butex woken by OnNewDataFromTcp, still need OnNewDataFromTcp.
+ if (options.user ==
static_cast<SocketUser*>(get_client_side_messenger())) {
+ _on_edge_trigger = rdma::RdmaEndpoint::OnNewDataFromTcp;
+ } else {
+ _on_edge_trigger = InputMessenger::OnNewMessages;
+ }
}
_tcp_transport = std::make_shared<TcpTransport>();
_tcp_transport->Init(socket, options);
@@ -78,7 +88,12 @@ std::shared_ptr<AppConnect> RdmaTransport::Connect() {
}
int RdmaTransport::CutFromIOBuf(butil::IOBuf *buf) {
- if (_rdma_ep && _rdma_state != RDMA_OFF) {
+ // Only send over the RDMA channel once the handshake has NEGOTIATED it
+ // (RDMA_ON). While the state is still RDMA_UNKNOWN (handshake in progress,
+ // or a server connection that turned out to be plain TCP and never
+ // handshook) or RDMA_OFF (fell back), the QP is not usable and everything
+ // must go over the TCP fd. Mirrors the RDMA_ON check in WaitEpollOut().
+ if (_rdma_ep && _rdma_state == RDMA_ON) {
butil::IOBuf *data_arr[1] = {buf};
return _rdma_ep->CutFromIOBufList(data_arr, 1);
} else {
@@ -87,7 +102,7 @@ int RdmaTransport::CutFromIOBuf(butil::IOBuf *buf) {
}
ssize_t RdmaTransport::CutFromIOBufList(butil::IOBuf **buf, size_t ndata) {
- if (_rdma_ep && _rdma_state != RDMA_OFF) {
+ if (_rdma_ep && _rdma_state == RDMA_ON) {
return _rdma_ep->CutFromIOBufList(buf, ndata);
}
return _tcp_transport->CutFromIOBufList(buf, ndata);
diff --git a/src/brpc/rdma_transport.h b/src/brpc/rdma_transport.h
index d8520b1a..0ac8a253 100644
--- a/src/brpc/rdma_transport.h
+++ b/src/brpc/rdma_transport.h
@@ -29,6 +29,7 @@ friend class TransportFactory;
friend class rdma::RdmaEndpoint;
friend class rdma::RdmaConnect;
friend class rdma::RdmaHandshakeServerV2;
+friend class rdma::RdmaHandshakeServerV3;
public:
void Init(Socket* socket, const SocketOptions& options) override;
void Release() override;
diff --git a/src/brpc/server.cpp b/src/brpc/server.cpp
index 99ecab8a..230c2284 100644
--- a/src/brpc/server.cpp
+++ b/src/brpc/server.cpp
@@ -607,13 +607,17 @@ int Server::AddBuiltinServices() {
return 0;
}
-bool is_http_protocol(const char* name) {
+BUTIL_FORCE_INLINE bool is_http_protocol(const char* name) {
if (name[0] != 'h') {
return false;
}
return strcmp(name, "http") == 0 || strcmp(name, "h2") == 0;
}
+BUTIL_FORCE_INLINE bool is_rdma_handshake_protocol(const char* name) {
+ return strcmp(name, "rdma_handshake") == 0;
+}
+
Acceptor* Server::BuildAcceptor() {
std::set<std::string> whitelist;
for (butil::StringSplitter sp(_options.enabled_protocols.c_str(), ' ');
@@ -637,6 +641,7 @@ Acceptor* Server::BuildAcceptor() {
}
if (has_whitelist &&
!is_http_protocol(protocols[i].name) &&
+ !is_rdma_handshake_protocol(protocols[i].name) &&
!whitelist.erase(protocols[i].name)) {
// the protocol is not allowed to serve.
RPC_VLOG << "Skip protocol=" << protocols[i].name;
diff --git a/src/brpc/socket.h b/src/brpc/socket.h
index ca83b921..b48c9bca 100644
--- a/src/brpc/socket.h
+++ b/src/brpc/socket.h
@@ -597,6 +597,9 @@ public:
// True if this socket was created by Connect.
bool CreatedByConnect() const;
+ // The socket mode decided at creation.
+ SocketMode socket_mode() const { return _socket_mode; }
+
// Get an UNUSED socket connecting to the same place as this socket
// from the SocketPool of this socket.
int GetPooledSocket(SocketUniquePtr* pooled_socket);
diff --git a/src/butil/raw_pack.h b/src/butil/raw_pack.h
index d4f8167f..f5f69f83 100644
--- a/src/butil/raw_pack.h
+++ b/src/butil/raw_pack.h
@@ -18,6 +18,7 @@
#ifndef BUTIL_RAW_PACK_H
#define BUTIL_RAW_PACK_H
+#include <string.h>
#include "butil/sys_byteorder.h"
namespace butil {
@@ -44,10 +45,16 @@ class RawPacker {
public:
// Notice: User must guarantee `stream' is as long as the packed data.
explicit RawPacker(void* stream) : _stream((char*)stream) {}
- ~RawPacker() {}
// Not using operator<< because some values may be packed differently from
// its type.
+ RawPacker& pack16(uint16_t host_value) {
+ *(uint16_t*)_stream = HostToNet16(host_value);
+ _stream += 2;
+ return *this;
+ }
+
+
RawPacker& pack32(uint32_t host_value) {
*(uint32_t*)_stream = HostToNet32(host_value);
_stream += 4;
@@ -62,6 +69,13 @@ public:
return *this;
}
+ // Pack `n' raw bytes from `data' as-is (no byte order conversion).
+ RawPacker& pack_bytes(const void* data, size_t n) {
+ memcpy(_stream, data, n);
+ _stream += n;
+ return *this;
+ }
+
private:
char* _stream;
};
@@ -71,21 +85,33 @@ private:
class RawUnpacker {
public:
explicit RawUnpacker(const void* stream) : _stream((const char*)stream) {}
- ~RawUnpacker() {}
- RawUnpacker& unpack32(uint32_t & host_value) {
+ RawUnpacker& unpack16(uint16_t& host_value) {
+ host_value = NetToHost16(*(const uint16_t*)_stream);
+ _stream += 2;
+ return *this;
+ }
+
+ RawUnpacker& unpack32(uint32_t& host_value) {
host_value = NetToHost32(*(const uint32_t*)_stream);
_stream += 4;
return *this;
}
- RawUnpacker& unpack64(uint64_t & host_value) {
+ RawUnpacker& unpack64(uint64_t& host_value) {
const uint32_t *p = (const uint32_t*)_stream;
host_value = (((uint64_t)NetToHost32(p[0])) << 32) | NetToHost32(p[1]);
_stream += 8;
return *this;
}
+ // Unpack `n' raw bytes into `data' as-is (no byte order conversion).
+ RawUnpacker& unpack_bytes(void* data, size_t n) {
+ memcpy(data, _stream, n);
+ _stream += n;
+ return *this;
+ }
+
private:
const char* _stream;
};
diff --git a/test/brpc_rdma_unittest.cpp b/test/brpc_rdma_unittest.cpp
index 43c6edfd..9aff49d3 100644
--- a/test/brpc_rdma_unittest.cpp
+++ b/test/brpc_rdma_unittest.cpp
@@ -39,6 +39,7 @@
#include "brpc/rdma/block_pool.h"
#include "brpc/rdma/rdma_endpoint.h"
#include "brpc/rdma/rdma_handshake.h"
+#include "brpc/rdma/rdma_handshake_constants.h"
#include "brpc/rdma/rdma_handshake.pb.h"
#include "brpc/rdma/rdma_helper.h"
#include "echo.pb.h"
@@ -55,8 +56,8 @@ DEFINE_bool(rdma_test_enable, false, "Enable tests requring
rdma runtime.");
namespace rdma {
-extern const uint16_t RDMA_HELLO_V2_VERSION;
-extern const uint16_t RDMA_IMPL_V2_VERSION;
+// HELLO_V2_VERSION / IMPL_V2_VERSION come from
+// brpc/rdma/rdma_handshake_constants.h (shared wire constants).
DECLARE_bool(rdma_trace_verbose);
DECLARE_int32(rdma_memory_pool_max_regions);
@@ -220,11 +221,14 @@ TEST_F(RdmaTest, client_hello_msg_invalid_magic_str) {
Socket* s = GetSocketFromServer(0);
ASSERT_EQ(rdma::RdmaEndpoint::UNINIT,
static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
memcpy(data, "PRPC", 4); // send as normal baidu_std protocol
ASSERT_EQ(4, write(sockfd, data, 4));
usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP,
static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ // A non-RDMA magic makes ParseRdmaHandshake return TRY_OTHERS and hand the
+ // bytes to other protocols; it does not touch the endpoint state, so it
+ // stays UNINIT (the old blocking handshake used to set FALLBACK_TCP here).
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT,
static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
StopServer();
}
@@ -248,7 +252,10 @@ TEST_F(RdmaTest, client_close_during_hello_send) {
memcpy(data, "RD", 2);
ASSERT_EQ(2, write(sockfd1, data, 2)); // break in magic str
usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_HELLO_WAIT,
static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ // Fewer than 4 magic bytes: ParseRdmaHandshake can't tell yet, returns
+ // NOT_ENOUGH_DATA and leaves the endpoint UNINIT (the old blocking
+ // handshake used to set S_HELLO_WAIT before reading the magic).
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT,
static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
close(sockfd1);
usleep(100000); // wait for server to handle the msg
ASSERT_EQ(NULL, GetSocketFromServer(0));
@@ -273,9 +280,13 @@ TEST_F(RdmaTest, client_close_during_hello_send) {
usleep(100000); // wait for server to handle the msg
s = GetSocketFromServer(0);
ASSERT_EQ(rdma::RdmaEndpoint::UNINIT,
static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ // Send the 4B magic plus a valid msg_len (=40) but no body, so the server
+ // recognizes an RDMA v2 hello and waits for the remaining bytes. (A zero
+ // msg_len would now be rejected up-front as a protocol error.)
memcpy(data, "RDMA", 4);
- memset(data + 4, 0, 4);
- ASSERT_EQ(8, write(sockfd3, data, 8)); // break after magic str
+ uint16_t v2_len = butil::HostToNet16(rdma::HELLO_V2_MSG_LEN_MIN);
+ memcpy(data + 4, &v2_len, sizeof(v2_len));
+ ASSERT_EQ(6, write(sockfd3, data, 6)); // magic + msg_len, body missing
usleep(100000); // wait for server to handle the msg
ASSERT_EQ(rdma::RdmaEndpoint::S_HELLO_WAIT,
static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
close(sockfd3);
@@ -293,7 +304,7 @@ TEST_F(RdmaTest, client_hello_msg_invalid_len) {
addr.sin_family = AF_INET;
addr.sin_port = htons(PORT);
Socket* s = NULL;
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
butil::fd_guard sockfd1(socket(AF_INET, SOCK_STREAM, 0));
ASSERT_TRUE(sockfd1 >= 0);
@@ -338,8 +349,8 @@ TEST_F(RdmaTest, client_hello_msg_invalid_version) {
addr.sin_family = AF_INET;
addr.sin_port = htons(PORT);
Socket* s = NULL;
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
- uint16_t len = butil::HostToNet16(rdma::v2_wire::HELLO_MSG_LEN_MIN);
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
+ uint16_t len = butil::HostToNet16(rdma::HELLO_V2_MSG_LEN_MIN);
uint16_t ver = butil::HostToNet16(1);
butil::fd_guard sockfd1(socket(AF_INET, SOCK_STREAM, 0));
@@ -359,7 +370,7 @@ TEST_F(RdmaTest, client_hello_msg_invalid_version) {
// UT mistakenly wrote `data, 36` which included the leftover "RDMA"
// magic at data[0..4); the server parsed it as msg_len = 0x5244 and
// happened to fall through to NegotiationValid (which then failed on
- // hello_ver). Now that Step 1 enforces a HELLO_MSG_LEN_MAX upper bound,
+ // hello_ver). Now that Step 1 enforces a HELLO_V2_MSG_LEN_MAX upper bound,
// such an oversized msg_len would be rejected before reaching the
// version check, breaking the intent of this UT.
ASSERT_EQ(36, write(sockfd1, data + 4, 36));
@@ -413,10 +424,10 @@ TEST_F(RdmaTest,
client_hello_msg_invalid_sq_rq_block_size) {
Socket* s = NULL;
uint32_t flags = butil::HostToNet32(0);
rdma::v2_wire::HelloMessage msg{};
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
- msg.msg_len = rdma::v2_wire::HELLO_MSG_LEN_MIN;
- msg.hello_ver = rdma::RDMA_HELLO_V2_VERSION;
- msg.impl_ver = rdma::RDMA_IMPL_V2_VERSION;
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
+ msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
+ msg.hello_ver = rdma::HELLO_V2_VERSION;
+ msg.impl_ver = rdma::IMPL_V2_VERSION;
msg.sq_size = 10;
msg.rq_size = 16;
@@ -505,10 +516,10 @@ TEST_F(RdmaTest, client_close_after_qp_build) {
addr.sin_port = htons(PORT);
Socket* s = NULL;
rdma::v2_wire::HelloMessage msg{};
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
- msg.msg_len = rdma::v2_wire::HELLO_MSG_LEN_MIN;
- msg.hello_ver = rdma::RDMA_HELLO_V2_VERSION;
- msg.impl_ver = rdma::RDMA_IMPL_V2_VERSION;
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
+ msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
+ msg.hello_ver = rdma::HELLO_V2_VERSION;
+ msg.impl_ver = rdma::IMPL_V2_VERSION;
msg.sq_size = 16;
msg.rq_size = 16;
msg.block_size = 8192;
@@ -542,10 +553,10 @@ TEST_F(RdmaTest, client_close_during_ack_send) {
addr.sin_port = htons(PORT);
Socket* s = NULL;
rdma::v2_wire::HelloMessage msg{};
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
- msg.msg_len = rdma::v2_wire::HELLO_MSG_LEN_MIN;
- msg.hello_ver = rdma::RDMA_HELLO_V2_VERSION;
- msg.impl_ver = rdma::RDMA_IMPL_V2_VERSION;
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
+ msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
+ msg.hello_ver = rdma::HELLO_V2_VERSION;
+ msg.impl_ver = rdma::IMPL_V2_VERSION;
msg.sq_size = 16;
msg.rq_size = 16;
msg.block_size = 8192;
@@ -586,10 +597,10 @@ TEST_F(RdmaTest, client_close_after_ack_send) {
addr.sin_port = htons(PORT);
Socket* s = NULL;
rdma::v2_wire::HelloMessage msg{};
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
- msg.msg_len = rdma::v2_wire::HELLO_MSG_LEN_MIN;
- msg.hello_ver = rdma::RDMA_HELLO_V2_VERSION;
- msg.impl_ver = rdma::RDMA_IMPL_V2_VERSION;
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
+ msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
+ msg.hello_ver = rdma::HELLO_V2_VERSION;
+ msg.impl_ver = rdma::IMPL_V2_VERSION;
msg.sq_size = 16;
msg.rq_size = 16;
msg.block_size = 8192;
@@ -651,10 +662,10 @@ TEST_F(RdmaTest, client_send_data_on_tcp_after_ack_send) {
addr.sin_port = htons(PORT);
Socket* s = NULL;
rdma::v2_wire::HelloMessage msg{};
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
- msg.msg_len = rdma::v2_wire::HELLO_MSG_LEN_MIN;
- msg.hello_ver = rdma::RDMA_HELLO_V2_VERSION;
- msg.impl_ver = rdma::RDMA_IMPL_V2_VERSION;
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
+ msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
+ msg.hello_ver = rdma::HELLO_V2_VERSION;
+ msg.impl_ver = rdma::IMPL_V2_VERSION;
msg.sq_size = 16;
msg.rq_size = 16;
msg.block_size = 8192;
@@ -763,8 +774,8 @@ TEST_F(RdmaTest, server_close_before_hello_send) {
butil::fd_guard acc_fd(accept(sockfd, NULL, NULL));
ASSERT_TRUE(acc_fd >= 0);
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, read(acc_fd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, read(acc_fd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
close(acc_fd);
usleep(100000);
ASSERT_EQ(rdma::RdmaEndpoint::FAILED,
static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
@@ -799,8 +810,8 @@ TEST_F(RdmaTest, server_miss_during_magic_str) {
butil::fd_guard acc_fd(accept(sockfd, NULL, NULL));
ASSERT_TRUE(acc_fd >= 0);
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, read(acc_fd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, read(acc_fd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
ASSERT_EQ(2, write(acc_fd, "RD", 2));
usleep(100000);
bthread_id_join(cntl.call_id());
@@ -834,8 +845,8 @@ TEST_F(RdmaTest, server_close_during_magic_str) {
butil::fd_guard acc_fd(accept(sockfd, NULL, NULL));
ASSERT_TRUE(acc_fd >= 0);
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, read(acc_fd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, read(acc_fd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
ASSERT_EQ(2, write(acc_fd, "RD", 2));
usleep(100000);
close(acc_fd);
@@ -872,8 +883,8 @@ TEST_F(RdmaTest, server_hello_invalid_magic_str) {
butil::fd_guard acc_fd(accept(sockfd, NULL, NULL));
ASSERT_TRUE(acc_fd >= 0);
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, read(acc_fd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, read(acc_fd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
ASSERT_EQ(4, write(acc_fd, "ABCD", 4));
usleep(100000);
ASSERT_EQ(rdma::RdmaEndpoint::FAILED,
static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
@@ -908,8 +919,8 @@ TEST_F(RdmaTest, server_miss_during_hello_msg) {
butil::fd_guard acc_fd(accept(sockfd, NULL, NULL));
ASSERT_TRUE(acc_fd >= 0);
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, read(acc_fd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, read(acc_fd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
ASSERT_EQ(4, write(acc_fd, "RDMA", 4));
ASSERT_EQ(2, write(acc_fd, "00", 2));
bthread_id_join(cntl.call_id());
@@ -943,8 +954,8 @@ TEST_F(RdmaTest, server_close_during_hello_msg) {
butil::fd_guard acc_fd(accept(sockfd, NULL, NULL));
ASSERT_TRUE(acc_fd >= 0);
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, read(acc_fd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, read(acc_fd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
ASSERT_EQ(4, write(acc_fd, "RDMA", 4));
ASSERT_EQ(2, write(acc_fd, "00", 2));
close(acc_fd);
@@ -981,13 +992,13 @@ TEST_F(RdmaTest, server_hello_invalid_msg_len) {
butil::fd_guard acc_fd(accept(sockfd, NULL, NULL));
ASSERT_TRUE(acc_fd >= 0);
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, read(acc_fd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, read(acc_fd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
memcpy(data, "RDMA", 4);
uint16_t len = butil::HostToNet16(35);
memcpy(data + 4, &len, 2);
memset(data + 6, 0, 32);
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, write(acc_fd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, write(acc_fd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
usleep(100000);
ASSERT_EQ(rdma::RdmaEndpoint::FAILED,
static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
bthread_id_join(cntl.call_id());
@@ -1021,13 +1032,13 @@ TEST_F(RdmaTest, server_hello_invalid_version) {
butil::fd_guard acc_fd(accept(sockfd, NULL, NULL));
ASSERT_TRUE(acc_fd >= 0);
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, read(acc_fd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, read(acc_fd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
memcpy(data, "RDMA", 4);
- uint16_t len = butil::HostToNet16(rdma::v2_wire::HELLO_MSG_LEN_MIN);
+ uint16_t len = butil::HostToNet16(rdma::HELLO_V2_MSG_LEN_MIN);
memcpy(data + 4, &len, 2);
memset(data + 6, 0, 32);
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, write(acc_fd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, write(acc_fd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
usleep(100000);
ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP,
static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
ASSERT_EQ(4, read(acc_fd, data, 4));
@@ -1064,11 +1075,11 @@ TEST_F(RdmaTest, server_hello_invalid_sq_rq_size) {
butil::fd_guard acc_fd(accept(sockfd, NULL, NULL));
ASSERT_TRUE(acc_fd >= 0);
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, read(acc_fd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, read(acc_fd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
rdma::v2_wire::HelloMessage msg{};
- msg.msg_len = rdma::v2_wire::HELLO_MSG_LEN_MIN;
+ msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
msg.hello_ver = 1;
msg.impl_ver = 1;
msg.sq_size = 0;
@@ -1078,7 +1089,7 @@ TEST_F(RdmaTest, server_hello_invalid_sq_rq_size) {
msg.gid = rdma::GetRdmaGid();
memcpy(data, "RDMA", 4);
msg.Serialize(data + 4);
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, write(acc_fd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, write(acc_fd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
usleep(100000);
ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP,
static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
@@ -1116,13 +1127,13 @@ TEST_F(RdmaTest, server_miss_after_ack) {
butil::fd_guard acc_fd(accept(sockfd, NULL, NULL));
ASSERT_TRUE(acc_fd >= 0);
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, read(acc_fd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, read(acc_fd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
rdma::v2_wire::HelloMessage msg{};
- msg.msg_len = rdma::v2_wire::HELLO_MSG_LEN_MIN;
- msg.hello_ver = rdma::RDMA_HELLO_V2_VERSION;
- msg.impl_ver = rdma::RDMA_IMPL_V2_VERSION;
+ msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
+ msg.hello_ver = rdma::HELLO_V2_VERSION;
+ msg.impl_ver = rdma::IMPL_V2_VERSION;
msg.sq_size = 16;
msg.rq_size = 16;
msg.block_size = 8192;
@@ -1130,7 +1141,7 @@ TEST_F(RdmaTest, server_miss_after_ack) {
msg.gid = rdma::GetRdmaGid();
memcpy(data, "RDMA", 4);
msg.Serialize(data + 4);
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, write(acc_fd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, write(acc_fd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
usleep(100000);
ASSERT_EQ(rdma::RdmaEndpoint::ESTABLISHED,
static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
@@ -1168,13 +1179,13 @@ TEST_F(RdmaTest, server_close_after_ack) {
butil::fd_guard acc_fd(accept(sockfd, NULL, NULL));
ASSERT_TRUE(acc_fd >= 0);
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, read(acc_fd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, read(acc_fd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
rdma::v2_wire::HelloMessage msg{};
- msg.msg_len = rdma::v2_wire::HELLO_MSG_LEN_MIN;
- msg.hello_ver = rdma::RDMA_HELLO_V2_VERSION;
- msg.impl_ver = rdma::RDMA_IMPL_V2_VERSION;
+ msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
+ msg.hello_ver = rdma::HELLO_V2_VERSION;
+ msg.impl_ver = rdma::IMPL_V2_VERSION;
msg.sq_size = 16;
msg.rq_size = 16;
msg.block_size = 8192;
@@ -1182,7 +1193,7 @@ TEST_F(RdmaTest, server_close_after_ack) {
msg.gid = rdma::GetRdmaGid();
memcpy(data, "RDMA", 4);
msg.Serialize(data + 4);
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, write(acc_fd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, write(acc_fd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
usleep(100000);
ASSERT_EQ(rdma::RdmaEndpoint::ESTABLISHED,
static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
@@ -1221,13 +1232,13 @@ TEST_F(RdmaTest, server_send_data_on_tcp_after_ack) {
butil::fd_guard acc_fd(accept(sockfd, NULL, NULL));
ASSERT_TRUE(acc_fd >= 0);
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, read(acc_fd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, read(acc_fd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
rdma::v2_wire::HelloMessage msg{};
- msg.msg_len = rdma::v2_wire::HELLO_MSG_LEN_MIN;
- msg.hello_ver = rdma::RDMA_HELLO_V2_VERSION;
- msg.impl_ver = rdma::RDMA_IMPL_V2_VERSION;
+ msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
+ msg.hello_ver = rdma::HELLO_V2_VERSION;
+ msg.impl_ver = rdma::IMPL_V2_VERSION;
msg.sq_size = 16;
msg.rq_size = 16;
msg.block_size = 8192;
@@ -1235,11 +1246,11 @@ TEST_F(RdmaTest, server_send_data_on_tcp_after_ack) {
msg.gid = rdma::GetRdmaGid();
memcpy(data, "RDMA", 4);
msg.Serialize(data + 4);
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, write(acc_fd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, write(acc_fd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
usleep(100000);
ASSERT_EQ(rdma::RdmaEndpoint::ESTABLISHED,
static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, write(acc_fd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, write(acc_fd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
bthread_id_join(cntl.call_id());
ASSERT_EQ(EPROTO, cntl.ErrorCode());
@@ -1272,26 +1283,26 @@ TEST_F(RdmaTest, v2_client_hello_bytes_baseline) {
butil::fd_guard acc_fd(accept(sockfd, NULL, NULL));
ASSERT_TRUE(acc_fd >= 0);
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, read(acc_fd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, read(acc_fd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
// [0..4) magic
ASSERT_EQ(0, memcmp(data, "RDMA", 4));
// [4..6) msg_len, big-endian uint16 == 40
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN,
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN,
(size_t)(((uint16_t)data[4] << 8) | (uint16_t)data[5]));
- // [6..8) hello_ver, big-endian uint16 == rdma::RDMA_HELLO_V2_VERSION
- ASSERT_EQ(rdma::RDMA_HELLO_V2_VERSION,
+ // [6..8) hello_ver, big-endian uint16 == rdma::HELLO_V2_VERSION
+ ASSERT_EQ(rdma::HELLO_V2_VERSION,
(uint16_t)(((uint16_t)data[6] << 8) | (uint16_t)data[7]));
- // [8..10) impl_ver, big-endian uint16 == rdma::RDMA_IMPL_V2_VERSION
- ASSERT_EQ(rdma::RDMA_IMPL_V2_VERSION,
+ // [8..10) impl_ver, big-endian uint16 == rdma::IMPL_V2_VERSION
+ ASSERT_EQ(rdma::IMPL_V2_VERSION,
(uint16_t)(((uint16_t)data[8] << 8) | (uint16_t)data[9]));
rdma::v2_wire::HelloMessage msg{};
msg.Deserialize(data + 4);
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, msg.msg_len);
- ASSERT_EQ(rdma::RDMA_HELLO_V2_VERSION, msg.hello_ver);
- ASSERT_EQ(rdma::RDMA_IMPL_V2_VERSION, msg.impl_ver);
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, msg.msg_len);
+ ASSERT_EQ(rdma::HELLO_V2_VERSION, msg.hello_ver);
+ ASSERT_EQ(rdma::IMPL_V2_VERSION, msg.impl_ver);
bthread_id_join(cntl.call_id());
}
@@ -1313,39 +1324,39 @@ TEST_F(RdmaTest, v2_server_hello_bytes_baseline) {
// Send a well-formed v2 hello so the server enters S_ACK_WAIT.
rdma::v2_wire::HelloMessage msg{};
- msg.msg_len = rdma::v2_wire::HELLO_MSG_LEN_MIN;
- msg.hello_ver = rdma::RDMA_HELLO_V2_VERSION;
- msg.impl_ver = rdma::RDMA_IMPL_V2_VERSION;
+ msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
+ msg.hello_ver = rdma::HELLO_V2_VERSION;
+ msg.impl_ver = rdma::IMPL_V2_VERSION;
msg.sq_size = 16;
msg.rq_size = 16;
msg.block_size = 8192;
msg.qp_num = 0;
msg.gid = rdma::GetRdmaGid();
- uint8_t data[rdma::v2_wire::HELLO_MSG_LEN_MIN];
+ uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
memcpy(data, "RDMA", 4);
msg.Serialize(data + 4);
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, write(sockfd, data,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, write(sockfd, data,
rdma::HELLO_V2_MSG_LEN_MIN));
usleep(100000);
ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT,
static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
// Read server's reply hello and assert its byte-level layout.
- uint8_t reply[rdma::v2_wire::HELLO_MSG_LEN_MIN];
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, read(sockfd, reply,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ uint8_t reply[rdma::HELLO_V2_MSG_LEN_MIN];
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, read(sockfd, reply,
rdma::HELLO_V2_MSG_LEN_MIN));
ASSERT_EQ(0, memcmp(reply, "RDMA", 4));
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN,
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN,
(size_t)(((uint16_t)reply[4] << 8) | (uint16_t)reply[5]));
- ASSERT_EQ(rdma::RDMA_HELLO_V2_VERSION,
+ ASSERT_EQ(rdma::HELLO_V2_VERSION,
(uint16_t)(((uint16_t)reply[6] << 8) | (uint16_t)reply[7]));
- ASSERT_EQ(rdma::RDMA_IMPL_V2_VERSION,
+ ASSERT_EQ(rdma::IMPL_V2_VERSION,
(uint16_t)(((uint16_t)reply[8] << 8) | (uint16_t)reply[9]));
rdma::v2_wire::HelloMessage reply_msg{};
reply_msg.Deserialize(reply + 4);
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, reply_msg.msg_len);
- ASSERT_EQ(rdma::RDMA_HELLO_V2_VERSION, reply_msg.hello_ver);
- ASSERT_EQ(rdma::RDMA_IMPL_V2_VERSION, reply_msg.impl_ver);
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, reply_msg.msg_len);
+ ASSERT_EQ(rdma::HELLO_V2_VERSION, reply_msg.hello_ver);
+ ASSERT_EQ(rdma::IMPL_V2_VERSION, reply_msg.impl_ver);
// Drive the server into FALLBACK_TCP via ACK flags=0 so the test ends
// cleanly without requiring real RDMA hardware.
@@ -1379,8 +1390,8 @@ TEST_F(RdmaTest, v2_server_drains_tail_then_reads_ack) {
// Build a v2 hello with msg_len = 48 (40 base + 8B zero tail).
rdma::v2_wire::HelloMessage msg{};
msg.msg_len = 48;
- msg.hello_ver = rdma::RDMA_HELLO_V2_VERSION;
- msg.impl_ver = rdma::RDMA_IMPL_V2_VERSION;
+ msg.hello_ver = rdma::HELLO_V2_VERSION;
+ msg.impl_ver = rdma::IMPL_V2_VERSION;
msg.sq_size = 16;
msg.rq_size = 16;
msg.block_size = 8192;
@@ -1423,23 +1434,23 @@ TEST_F(RdmaTest, v2_server_rejects_oversized_msg_len) {
ASSERT_TRUE(s != NULL);
ASSERT_EQ(rdma::RdmaEndpoint::UNINIT,
static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- // Build a v2 hello with msg_len = 4097 (HELLO_MSG_LEN_MAX + 1).
+ // Build a v2 hello with msg_len = 4097 (HELLO_V2_MSG_LEN_MAX + 1).
// We only send the 40B base; the server must reject before reading
// (and definitely before attempting to drain) any "tail".
rdma::v2_wire::HelloMessage msg{};
msg.msg_len = 4097;
- msg.hello_ver = rdma::RDMA_HELLO_V2_VERSION;
- msg.impl_ver = rdma::RDMA_IMPL_V2_VERSION;
+ msg.hello_ver = rdma::HELLO_V2_VERSION;
+ msg.impl_ver = rdma::IMPL_V2_VERSION;
msg.sq_size = 16;
msg.rq_size = 16;
msg.block_size = 8192;
msg.qp_num = 0;
msg.gid = rdma::GetRdmaGid();
- uint8_t buf[rdma::v2_wire::HELLO_MSG_LEN_MIN];
+ uint8_t buf[rdma::HELLO_V2_MSG_LEN_MIN];
memcpy(buf, "RDMA", 4);
msg.Serialize(buf + 4);
- ASSERT_EQ(rdma::v2_wire::HELLO_MSG_LEN_MIN, write(sockfd, buf,
rdma::v2_wire::HELLO_MSG_LEN_MIN));
+ ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, write(sockfd, buf,
rdma::HELLO_V2_MSG_LEN_MIN));
usleep(100000);
@@ -1659,7 +1670,9 @@ TEST_F(RdmaTest, v3_server_rejects_oversized_pb_size) {
uint8_t buf[8];
memcpy(buf, "RDM3", 4);
- uint32_t pb_size_be = butil::HostToNet32(4097);
+ // pb_size just above the allowed maximum -> rejected.
+ uint32_t pb_size_be =
+ butil::HostToNet32(static_cast<uint32_t>(rdma::HELLO_V3_MAX_PB_SIZE +
1));
memcpy(buf + 4, &pb_size_be, 4);
ASSERT_EQ(8, write(sockfd, buf, 8));
usleep(100000);
@@ -1958,7 +1971,7 @@ TEST_P(RdmaRpcTest, rdma_client_to_tcp_server) {
::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
usleep(100000);
bthread_id_join(cntl.call_id());
- ASSERT_EQ(EEOF, cntl.ErrorCode());
+ ASSERT_FALSE(cntl.Failed());
StopServer();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
