chenBright commented on code in PR #3535:
URL: https://github.com/apache/brpc/pull/3535#discussion_r4198275305
##########
src/brpc/global.cpp:
##########
@@ -438,12 +438,16 @@ static void GlobalInitializeOrDieImpl() {
}
// Protocols
- Protocol rdma_handshake_protocol = {
- ParseRdmaHandshake, nullptr, nullptr,
- ProcessRdmaHandshake, nullptr,
+ Protocol transport_handshake_protocol = {
+ ParseTransportHandshake, nullptr, nullptr,
+ ProcessTransportHandshake, nullptr,
nullptr, nullptr, nullptr,
CONNECTION_TYPE_ALL, "rdma_handshake" };
- if (RegisterProtocol(PROTOCOL_RDMA_HANDSHAKE, rdma_handshake_protocol) !=
0) {
+ // Retain the existing enum value and registered name to avoid changing
+ // public protocol identifiers while widening the implementation from RDMA
+ // to all transport upgrades.
+ if (RegisterProtocol(PROTOCOL_RDMA_HANDSHAKE,
Review Comment:
I think modifying `rdma_handshake` and `PROTOCOL_RDMA_HANDSHAKE` is fine.
##########
src/brpc/handshake/handshake_adapter.cpp:
##########
@@ -0,0 +1,55 @@
+// 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/handshake/handshake_adapter.h"
+
+#include "brpc/socket.h"
+
+namespace brpc {
+namespace handshake {
+
+// InputMessenger may call this entry repeatedly while bytes arrive. Keep all
+// connection state in HandshakeSession and Socket rather than in the adapter,
+// so a single stateless adapter can serve every connection. The parsing
+// context retains that selected adapter between the server hello and the peer
+// ACK, whose frame has no protocol magic of its own.
+ParseResult StandardHandshakeAdapter::ExecuteServerHandshake(
+ butil::IOBuf* source, Socket* socket) {
+ const StepResult result = RunServerStep(source, socket);
+ if (result == STEP_NEED_MORE) {
+ if (GetSession(socket)->phase() == ACK_WAIT &&
+ socket->parsing_context() == NULL) {
Review Comment:
Use nullptr instead of NULL .
##########
src/brpc/rdma_transport.cpp:
##########
@@ -30,27 +29,26 @@ DECLARE_bool(usercode_in_pthread);
extern SocketVarsCollector *g_vars;
+RdmaTransport *RdmaTransport::Get(const Socket *socket) {
+ const AdapterTransport *adapter = AdapterTransport::Get(socket);
Review Comment:
Same as above.
##########
src/brpc/rdma_transport.cpp:
##########
@@ -30,27 +29,26 @@ DECLARE_bool(usercode_in_pthread);
extern SocketVarsCollector *g_vars;
+RdmaTransport *RdmaTransport::Get(const Socket *socket) {
Review Comment:
`RdmaTransport *` -> `RdmaTransport* `
##########
src/brpc/rdma_transport.cpp:
##########
@@ -30,27 +29,26 @@ DECLARE_bool(usercode_in_pthread);
extern SocketVarsCollector *g_vars;
+RdmaTransport *RdmaTransport::Get(const Socket *socket) {
+ const AdapterTransport *adapter = AdapterTransport::Get(socket);
+ Transport *transport = adapter->high_speed_transport();
+ CHECK(transport != NULL);
+ return static_cast<RdmaTransport *>(transport);
Review Comment:
`static_cast` seems unnecessary.
##########
src/brpc/adapter_transport.h:
##########
@@ -0,0 +1,101 @@
+// 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_ADAPTER_TRANSPORT_H
+#define BRPC_ADAPTER_TRANSPORT_H
+
+#include <memory>
+
+#include "brpc/socket_mode.h"
+#include "brpc/transport.h"
+#include "brpc/transport_handshake.h"
+#include "brpc/parse_result.h"
+
+namespace brpc {
+
+class TcpTransport;
+class RdmaTransport;
+class UBShmTransport;
+
+// The top-level Transport installed in Socket. It starts on TcpTransport and
+// may switch to an independent RDMA/URMA/UBSHM Transport after a successful
+// handshake. TCP remains usable before negotiation and after fallback.
+class AdapterTransport : public Transport {
+ friend class TransportFactory;
+ friend class RdmaTransport;
+ friend class UBShmTransport;
Review Comment:
No indentation required.
##########
src/brpc/adapter_transport.cpp:
##########
@@ -0,0 +1,673 @@
+// 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/adapter_transport.h"
+
+#include <algorithm>
+#include <cstdint>
+#include <cstring>
+#include <errno.h>
+#include <unistd.h>
+
+#include "brpc/input_messenger.h"
+#include "brpc/destroyable.h"
+#include "brpc/handshake/rdma_handshake.h"
+#include "brpc/handshake/ubshm_handshake.h"
+#if BRPC_WITH_RDMA
+#include "brpc/rdma/rdma_helper.h"
+#endif
+#if BRPC_WITH_UBRING
+#include "brpc/ubshm/ub_endpoint.h"
+#include "brpc/ubshm/ub_helper.h"
+#include "brpc/ubshm/ubr_trx.h"
+#endif
+#include "brpc/rdma_transport.h"
+#include "brpc/tcp_transport.h"
+#include "brpc/ubshm_transport.h"
+
+namespace brpc {
+
+namespace {
+
+bool MatchesMagicPrefix(const char *prefix, size_t prefix_len,
+ const char *magic, size_t magic_len) {
+ const size_t compare_len = std::min(prefix_len, magic_len);
+ return memcmp(prefix, magic, compare_len) == 0;
+}
+
+class AdapterConnect : public AppConnect {
+public:
+ explicit AdapterConnect(const std::shared_ptr<AppConnect>& app_connect)
+ : _app_connect(app_connect) {}
+
+ static std::shared_ptr<AppConnect> Wrap(
+ const std::shared_ptr<AppConnect>& app_connect) {
+ if (std::dynamic_pointer_cast<AdapterConnect>(app_connect)) {
+ return app_connect;
+ }
+ return std::make_shared<AdapterConnect>(app_connect);
+ }
+
+ static std::shared_ptr<AppConnect> Unwrap(
+ const std::shared_ptr<AppConnect>& app_connect) {
+ const std::shared_ptr<AdapterConnect> adapter =
+ std::dynamic_pointer_cast<AdapterConnect>(app_connect);
+ return adapter ? adapter->_app_connect : app_connect;
+ }
+
+ void StartConnect(const Socket* socket,
+ void (*done)(int, void*), void* data) override {
+ ApplicationConnectTask* task = new ApplicationConnectTask{
+ socket, _app_connect, done, data};
+ if (AdapterTransport::StartClientUpgrade(
+ socket, OnUpgradeComplete, task) != 0) {
+
AdapterTransport::Get(const_cast<Socket*>(socket))->CompleteConnection(
+ handshake::FAILED);
+ const int error = errno != 0 ? errno : EAGAIN;
+ delete task;
+ done(error, data);
+ }
+ }
+
+ void StopConnect(Socket*) override {}
+
+private:
+ struct ApplicationConnectTask {
+ const Socket* socket;
+ std::shared_ptr<AppConnect> app_connect;
+ void (*done)(int, void*);
+ void* data;
+ };
+
+ static void OnApplicationComplete(int error, void* arg) {
+ std::unique_ptr<ApplicationConnectTask> task(
+ static_cast<ApplicationConnectTask*>(arg));
+ task->done(error, task->data);
+ }
+
+ static void OnUpgradeComplete(int error, void* arg) {
+ ApplicationConnectTask* task =
+ static_cast<ApplicationConnectTask*>(arg);
+ if (error != 0 || !task->app_connect) {
+ std::unique_ptr<ApplicationConnectTask> owned(task);
+ task->done(error, task->data);
+ return;
+ }
+ task->app_connect->StartConnect(
+ task->socket, OnApplicationComplete, task);
+ }
+
+ std::shared_ptr<AppConnect> _app_connect;
+};
+
+struct ClientHandshakeTask {
+ AdapterTransport* adapter;
+ void (*done)(int, void*);
+ void* data;
+ SocketUniquePtr socket;
+};
+
+#if BRPC_WITH_RDMA
+class RdmaClientHandshakeTransport : public handshake::HandshakeTransport {
+public:
+ RdmaClientHandshakeTransport(
+ RdmaTransport* transport, rdma::RdmaHandshakeAdapter* protocol,
+ Socket* socket, int* connect_error)
+ : _transport(transport), _protocol(protocol), _socket(socket),
+ _connect_error(connect_error) {}
+
+ handshake::StepResult PrepareResources() override {
+ if (_transport->PrepareUpgradeResources() == 0) {
+ return handshake::STEP_OK;
+ }
+ errno = 0;
+ return handshake::STEP_FALLBACK;
+ }
+
+ handshake::StepResult NegotiateResources() override {
+ return _transport->NegotiateUpgradeResources(
+ _protocol->remote(), false) == 0
+ ? handshake::STEP_OK : handshake::STEP_FALLBACK;
+ }
+
+ void OnEstablished() override { _transport->ActivateUpgrade(); }
+
+ void OnFallback() override { _transport->DeactivateUpgrade(); }
+
+ void OnFailed() override {
+ _transport->DeactivateUpgrade();
+ const int saved_errno = errno != 0 ? errno : EPROTO;
+ *_connect_error = saved_errno;
+ _socket->SetFailed(saved_errno,
+ "Fail to complete rdma handshake from %s: %s",
+ _socket->description().c_str(),
+ berror(saved_errno));
+ }
+
+private:
+ RdmaTransport* _transport;
+ rdma::RdmaHandshakeAdapter* _protocol;
+ Socket* _socket;
+ int* _connect_error;
+};
+#endif
+
+#if BRPC_WITH_UBRING
+class UBShmClientHandshakeTransport : public handshake::HandshakeTransport {
+public:
+ UBShmClientHandshakeTransport(
+ UBShmTransport* transport, ubring::SHM* local_shm,
+ const std::string& shm_name, Socket* socket, int* connect_error)
+ : _transport(transport), _local_shm(local_shm),
+ _shm_name(shm_name), _socket(socket),
+ _connect_error(connect_error) {}
+
+ handshake::StepResult PrepareResources() override {
+ return _transport->PrepareUpgradeResources(
+ _local_shm, _shm_name.c_str()) == 0
+ ? handshake::STEP_OK : handshake::STEP_FALLBACK;
+ }
+
+ handshake::StepResult NegotiateResources() override {
+ return _transport->NegotiateUpgradeResources(
+ _local_shm, _shm_name.c_str()) == 0
+ ? handshake::STEP_OK : handshake::STEP_FALLBACK;
+ }
+
+ void OnEstablished() override { _transport->ActivateUpgrade(); }
+ void OnFallback() override { _transport->DeactivateUpgrade(); }
+
+ void OnFailed() override {
+ _transport->DeactivateUpgrade();
+ const int saved_errno = errno != 0 ? errno : EPROTO;
+ *_connect_error = saved_errno;
+ _socket->SetFailed(saved_errno,
+ "Fail to complete ubring handshake from %s: %s",
+ _socket->description().c_str(),
+ berror(saved_errno));
+ }
+
+private:
+ UBShmTransport* _transport;
+ ubring::SHM* _local_shm;
+ std::string _shm_name;
+ Socket* _socket;
+ int* _connect_error;
+};
+#endif
+
+} // namespace
+
+AdapterTransport::AdapterTransport(SocketMode mode)
+ : _mode(mode), _connection_completed(0) {}
+AdapterTransport::~AdapterTransport() = default;
+
+AdapterTransport* AdapterTransport::Get(Socket* socket) {
+ CHECK(socket != NULL);
+ return static_cast<AdapterTransport*>(socket->_transport.get());
+}
+
+const AdapterTransport* AdapterTransport::Get(const Socket* socket) {
+ CHECK(socket != NULL);
+ return static_cast<const AdapterTransport*>(socket->_transport.get());
+}
+
+bool AdapterTransport::upgrade_capable(SocketMode mode) const {
+ if (_mode != mode || _high_speed_transport == NULL) {
+ return false;
+ }
+ switch (mode) {
+#if BRPC_WITH_RDMA
+ case SOCKET_MODE_RDMA:
+ return static_cast<RdmaTransport*>(
+ _high_speed_transport.get())->UpgradeReady();
+#endif
+#if BRPC_WITH_UBRING
+ case SOCKET_MODE_UBRING:
+ return static_cast<UBShmTransport*>(
+ _high_speed_transport.get())->UpgradeReady();
+#endif
+ default:
+ return false;
+ }
+}
+
+int AdapterTransport::StartClientUpgrade(const Socket* socket,
+ void (*done)(int, void*),
+ void* data) {
+ AdapterTransport* adapter = Get(const_cast<Socket*>(socket));
+ ClientHandshakeTask* task = new ClientHandshakeTask{adapter, done, data,
SocketUniquePtr()};
+ if (Socket::Address(socket->id(), &task->socket) != 0) {
+ delete task;
+ return -1;
+ }
+ bthread_t tid;
+ bthread_attr_t attr = BTHREAD_ATTR_NORMAL;
+ bthread_attr_set_name(&attr, "StartClientUpgrade");
+ if (bthread_start_background(&tid, &attr,
+ ProcessClientHandshake, task) < 0) {
+ delete task;
+ return -1;
+ }
+ return 0;
+}
+
+ParseResult AdapterTransport::ProcessUpgradeReadable(butil::IOBuf* source) {
+ ParseResult result(PARSE_ERROR_NOT_ENOUGH_DATA);
+ if (_socket->parsing_context() != NULL) {
+ handshake::ServerHandshakeContext* context =
+ static_cast<handshake::ServerHandshakeContext*>(
+ _socket->parsing_context());
+ CHECK(context->adapter() != NULL);
+ result = context->adapter()->ExecuteServerHandshake(source, _socket);
+ } else if (!source->empty()) {
+ static const size_t MAX_MAGIC_LEN = 4;
+ char prefix[MAX_MAGIC_LEN] = {};
+ const size_t prefix_len = std::min(source->size(), MAX_MAGIC_LEN);
+ source->copy_to(prefix, prefix_len);
+
+ const bool matches_ub =
+ MatchesMagicPrefix(prefix, prefix_len, "UB", 2);
+ const bool matches_rdma =
+ MatchesMagicPrefix(prefix, prefix_len, "RDMA", 4) ||
+ MatchesMagicPrefix(prefix, prefix_len, "RDM3", 4);
+ if (!matches_ub && !matches_rdma) {
+ result = ParseResult(PARSE_ERROR_TRY_OTHERS);
+ } else {
+ handshake::HandshakeAdapter* adapter =
+ matches_ub
+ ? handshake::GetUBShmServerHandshakeAdapter()
+ : handshake::GetRdmaServerHandshakeAdapter();
+ result = adapter->ExecuteServerHandshake(source, _socket);
+ }
+ }
+ const int phase = _handshake.phase();
+ if (!connection_completed() &&
+ (phase == handshake::ESTABLISHED ||
+ phase == handshake::FALLBACK_TCP || phase == handshake::FAILED)) {
+ CompleteConnection(static_cast<handshake::Phase>(phase));
+ }
+ return result;
+}
+
+void AdapterTransport::CompleteConnection(handshake::Phase terminal_phase) {
+ CHECK(terminal_phase == handshake::ESTABLISHED ||
+ terminal_phase == handshake::FALLBACK_TCP ||
+ terminal_phase == handshake::FAILED);
+ if (terminal_phase == handshake::FAILED &&
+ _handshake.phase() != handshake::FAILED) {
+ _handshake.MarkFailed();
+ }
+ int expected = 0;
+ _connection_completed.compare_exchange_strong(
+ expected, 1, butil::memory_order_release,
+ butil::memory_order_relaxed);
+}
+
+void* AdapterTransport::ProcessClientHandshake(void* arg) {
+ std::unique_ptr<ClientHandshakeTask> task(
+ static_cast<ClientHandshakeTask*>(arg));
+ AdapterTransport* adapter = task->adapter;
+ Socket* socket = task->socket.get();
+ int connect_error = 0;
+ (void)connect_error;
+
+#if BRPC_WITH_RDMA
+ if (adapter->_mode == SOCKET_MODE_RDMA) {
+ RdmaTransport* transport = static_cast<RdmaTransport*>(
+ adapter->_high_speed_transport.get());
+ if (!rdma::IsRdmaAvailable()) {
+ adapter->FallbackToTcp();
+ adapter->CompleteConnection(handshake::FALLBACK_TCP);
+ task->done(0, task->data);
+ return NULL;
+ }
+
+ std::unique_ptr<rdma::RdmaHandshakeAdapter> protocol =
+ transport->CreateClientHandshakeAdapter();
+ CHECK(protocol != NULL);
+ RdmaClientHandshakeTransport participant(
+ transport, protocol.get(), socket, &connect_error);
+ const handshake::StepResult result = adapter->_handshake.RunClient(
+ protocol.get(), &participant);
+ if (result == handshake::STEP_OK &&
+ transport->StartUpgradeEvents() < 0) {
+ const int saved_errno = errno != 0 ? errno : ERDMA;
+ transport->DeactivateUpgrade();
+ adapter->_handshake.MarkFailed();
+ socket->SetFailed(
+ saved_errno,
+ "Fail to start RDMA CQ events from %s: %s",
+ socket->description().c_str(), berror(saved_errno));
+ connect_error = saved_errno;
+ }
+ if (result == handshake::STEP_ERROR && connect_error == 0) {
+ connect_error = errno != 0 ? errno : EPROTO;
+ }
+ adapter->CompleteConnection(static_cast<handshake::Phase>(
+ adapter->_handshake.phase()));
+ task->done(connect_error, task->data);
+ return NULL;
+ }
+#endif
+
+#if BRPC_WITH_UBRING
+ if (adapter->_mode == SOCKET_MODE_UBRING) {
+ UBShmTransport* transport = static_cast<UBShmTransport*>(
+ adapter->_high_speed_transport.get());
+ if (!ubring::IsUBAvailable()) {
+ adapter->FallbackToTcp();
+ adapter->CompleteConnection(handshake::FALLBACK_TCP);
+ task->done(0, task->data);
+ return NULL;
+ }
+
+ const size_t local_shm_len =
+ static_cast<size_t>(ubring::FLAGS_data_queue_size) * MB_TO_BYTE;
+ ubring::SHM local_trx_shm = {
+ NULL, local_shm_len, 0, {0}, static_cast<uint32_t>(socket->fd())};
+ const auto shm_name_str =
+ butil::endpoint2str(socket->local_side());
+ ubring::UBShmHandshakeAdapter wire;
+ wire.ConfigureClientHello(local_shm_len, shm_name_str.c_str());
Review Comment:
`ConfigureClientHello()` copies the client's address string before resource
allocation. However, `ApplyAndMapLocalShm()` creates the shared-memory object
using the name `UBRING_<client-address>_C` and stores that name in
`local_trx_shm.name`.
The Hello therefore advertises a different name from the allocated object.
The server uses the advertised name directly when mapping the client's shared
memory, causing resource preparation to fail and the connection to fall back to
TCP.
##########
test/brpc_rdma_unittest.cpp:
##########
@@ -62,24 +59,26 @@ DEFINE_bool(rdma_test_enable, false, "Enable tests requring
rdma runtime.");
namespace rdma {
// HELLO_V2_VERSION / IMPL_V2_VERSION come from
-// brpc/rdma/rdma_handshake_constants.h (shared wire constants).
+// brpc/handshake/rdma_handshake_constants.h (shared wire constants).
DECLARE_bool(rdma_trace_verbose);
DECLARE_int32(rdma_memory_pool_max_regions);
DECLARE_int32(rdma_client_handshake_version);
DECLARE_bool(rdma_ece);
-extern ibv_cq* (*IbvCreateCq)(ibv_context*, int, void*, ibv_comp_channel*,
int);
+extern ibv_cq* (*IbvCreateCq)(ibv_context*, int, void*, ibv_comp_channel*,
+ int);
extern int (*IbvDestroyCq)(ibv_cq*);
extern ibv_qp* (*IbvCreateQp)(ibv_pd*, ibv_qp_init_attr*);
extern int (*IbvModifyQp)(ibv_qp*, ibv_qp_attr*, ibv_qp_attr_mask);
-extern int (*IbvQueryQp)(ibv_qp*, ibv_qp_attr*, ibv_qp_attr_mask,
ibv_qp_init_attr*);
+extern int (*IbvQueryQp)(ibv_qp*, ibv_qp_attr*, ibv_qp_attr_mask,
+ ibv_qp_init_attr*);
Review Comment:
No need to modify.
##########
test/brpc_rdma_unittest.cpp:
##########
@@ -376,144 +358,192 @@ TEST_F(RdmaTest,
stale_cq_callback_does_not_poll_new_generation) {
TEST_F(RdmaTest, client_close_before_hello_send) {
StartServer();
- butil::fd_guard sockfd;
- ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd));
- Socket* s = WaitForServerSocket();
- ASSERT_TRUE(s != nullptr);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT,
RdmaTransportOf(s)->_rdma_ep->_state);
- sockfd.reset(-1);
- ASSERT_TRUE(WaitForServerSocketGone());
+ sockaddr_in addr;
+ bzero((char*)&addr, sizeof(addr));
+ addr.sin_family = AF_INET;
+ addr.sin_port = htons(PORT);
+
+ butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0));
+ ASSERT_TRUE(sockfd >= 0);
+ ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
+ usleep(100000); // wait for server to handle the msg
+ Socket* s = GetSocketFromServer(0);
+ ASSERT_EQ(handshake::UNINITIALIZED,
+ AdapterTransport::Get(s)->handshake_phase());
+ close(sockfd);
+ usleep(100000); // wait for server to handle the msg
+ ASSERT_EQ(nullptr, GetSocketFromServer(0));
StopServer();
}
TEST_F(RdmaTest, client_hello_msg_invalid_magic_str) {
StartServer();
- butil::fd_guard sockfd;
- ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd));
- Socket* s = WaitForServerSocket();
- ASSERT_TRUE(s != nullptr);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT,
RdmaTransportOf(s)->_rdma_ep->_state);
+ sockaddr_in addr;
+ bzero((char*)&addr, sizeof(addr));
+ addr.sin_family = AF_INET;
+ addr.sin_port = htons(PORT);
+
+ butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0));
+ ASSERT_TRUE(sockfd >= 0);
+ ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
+ usleep(100000); // wait for server to handle the msg
+ Socket* s = GetSocketFromServer(0);
+ ASSERT_EQ(handshake::UNINITIALIZED,
+ AdapterTransport::Get(s)->handshake_phase());
uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
memcpy(data, "PRPC", 4); // send as normal baidu_std protocol
- ASSERT_TRUE(WriteAll(sockfd, data, 4));
- // Wait for the bytes to show up in the fd stream (baidu_std wants 12B of
- // header, so they stay buffered). Waiting on the state instead would prove
- // nothing: it is already UNINIT before the server has read anything.
- ASSERT_TRUE(WaitForFdReadBuf(s, 4));
- // 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,
RdmaTransportOf(s)->_rdma_ep->_state);
+ ASSERT_EQ(4, write(sockfd, data, 4));
+ usleep(100000); // wait for server to handle the msg
+ // A non-RDMA magic makes the transport-handshake parser 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(handshake::UNINITIALIZED,
+ AdapterTransport::Get(s)->handshake_phase());
StopServer();
}
TEST_F(RdmaTest, client_close_during_hello_send) {
StartServer();
+ sockaddr_in addr;
+ bzero((char*)&addr, sizeof(addr));
+ addr.sin_family = AF_INET;
+ addr.sin_port = htons(PORT);
Socket* s = nullptr;
uint8_t data[8];
- butil::fd_guard sockfd1;
- ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd1));
- s = WaitForServerSocket();
- ASSERT_TRUE(s != nullptr);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT,
RdmaTransportOf(s)->_rdma_ep->_state);
+ butil::fd_guard sockfd1(socket(AF_INET, SOCK_STREAM, 0));
+ ASSERT_TRUE(sockfd1 >= 0);
+ ASSERT_EQ(0, connect(sockfd1, (sockaddr*)&addr, sizeof(sockaddr)));
+ usleep(100000); // wait for server to handle the msg
Review Comment:
Do not use sleep to ensure correctness. I think it's fine to stick with the
original usage.
##########
src/brpc/rdma_transport.cpp:
##########
@@ -30,27 +29,26 @@ DECLARE_bool(usercode_in_pthread);
extern SocketVarsCollector *g_vars;
+RdmaTransport *RdmaTransport::Get(const Socket *socket) {
+ const AdapterTransport *adapter = AdapterTransport::Get(socket);
+ Transport *transport = adapter->high_speed_transport();
Review Comment:
Same as above.
##########
src/brpc/transport_handshake.h:
##########
@@ -0,0 +1,233 @@
+// 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_TRANSPORT_HANDSHAKE_H
+#define BRPC_TRANSPORT_HANDSHAKE_H
+
+#include <cstddef>
+#include <string>
+#include <vector>
+
+#include "butil/atomicops.h"
+#include "butil/macros.h"
+#include "brpc/destroyable.h"
+#include "brpc/handshake/handshake_frame.h"
+
+namespace brpc {
+
+class Socket;
+
+namespace handshake {
+
+class HandshakeAdapter;
+
+// Context retained by InputMessenger between the hello and ACK parse calls.
+// Remembering the selected stateless adapter is necessary because ACK frames
+// have no magic and cannot be dispatched from their bytes alone.
+struct ServerHandshakeContext : public Destroyable {
+ ServerHandshakeContext() : _adapter(NULL) {}
+ static ServerHandshakeContext* Create(HandshakeAdapter* adapter);
+ HandshakeAdapter* adapter() const { return _adapter; }
+ void Destroy() override;
+
+private:
+ HandshakeAdapter* _adapter;
+};
+
+// Protocol adapters may use transport-specific intermediate values, but the
+// terminal values are shared so that AdapterTransport can make the same
+// acquire-side decision for RDMA, URMA and UBSHM.
+enum Phase {
+ UNINITIALIZED = 0,
+ PREPARING = 1,
+ HELLO_SEND = 2,
+ HELLO_WAIT = 3,
+ NEGOTIATING = 4,
+ ACK_SEND = 5,
+ ACK_WAIT = 6,
+ EXTENSION_SEND = 7,
+ EXTENSION_WAIT = 8,
+ ESTABLISHED = 0x100,
+ FALLBACK_TCP = 0x200,
+ FAILED = 0x300,
+};
+
+enum StepResult {
+ STEP_OK = 0,
+ STEP_FALLBACK,
+ STEP_NEED_MORE,
+ STEP_NOT_MINE,
+ STEP_ERROR,
+};
+
+// Wire-level participant in a transport upgrade. Implementations own parsed
+// protocol state; HandshakeSession owns framing and phase orchestration.
+class HandshakeProtocol {
+public:
+ virtual ~HandshakeProtocol() = default;
+
+ virtual int ProtocolVersion() const = 0;
+ virtual const FrameSpec& HelloFrameSpec() const = 0;
+ virtual const FrameSpec& AckFrameSpec() const = 0;
+ virtual StepResult BuildHello(bool enabled, std::string* payload) = 0;
+ virtual StepResult ParseHello(const std::string& payload) = 0;
+
+ // RDMA and UBSHM use the same four-byte, network-order ACK. Protocols
+ // with a different ACK format may override these methods.
+ virtual StepResult BuildAck(bool enabled, std::string* payload);
+ virtual StepResult ParseAck(const std::string& payload, bool* enabled);
+
+ virtual bool HasExtension() const { return false; }
+ virtual const FrameSpec& ExtensionFrameSpec() const;
+ virtual StepResult BuildExtension(bool, std::string*) {
+ return STEP_ERROR;
+ }
+ virtual StepResult ParseExtension(const std::string&) {
+ return STEP_ERROR;
+ }
+};
+
+// Resource-level participant in a transport upgrade. Cleanup remains part of
+// this contract: resources may already exist when negotiation falls back or
+// a framing/I/O error terminates the handshake.
+class HandshakeTransport {
+public:
+ virtual ~HandshakeTransport() = default;
+
+ virtual void OnProtocolSelected(HandshakeProtocol*) {}
+ virtual StepResult PrepareResources() = 0;
+ virtual StepResult NegotiateResources() = 0;
+ virtual void OnEstablished() = 0;
+ virtual void OnFallback() = 0;
+ virtual void OnFailed() = 0;
+ virtual StepResult ValidateEstablished() { return STEP_OK; }
+};
+
+// Participant for a server that recognizes an upgrade protocol only to
+// negotiate TCP fallback. It owns no high-speed resources.
+class FallbackHandshakeTransport : public HandshakeTransport {
+public:
+ StepResult PrepareResources() override { return STEP_OK; }
+ StepResult NegotiateResources() override { return STEP_OK; }
+ void OnEstablished() override {}
+ void OnFallback() override {}
+ void OnFailed() override {}
+};
+
+class FallbackHandshakeProtocol : public HandshakeProtocol {
+public:
+ StepResult ParseHello(const std::string&) override {
+ return STEP_FALLBACK;
+ }
+ StepResult ParseAck(const std::string& payload, bool* enabled) override {
+ bool ignored = false;
+ const StepResult result = HandshakeProtocol::ParseAck(
+ payload, &ignored);
+ *enabled = false;
+ return result;
+ }
+};
+
+// Owns one connection-upgrade attempt, invokes the protocol field codec and
+// resource callbacks, and provides common framing, TCP control-plane I/O,
+// lifecycle and publication ordering.
+class HandshakeSession {
+public:
+ explicit HandshakeSession(Socket* socket = NULL)
+ : _socket_io(socket), _io(&_socket_io), _phase(UNINITIALIZED),
+ _protocol_version(0), _local_enabled(false) {}
+
+ void Reset(Socket* socket) {
+ _socket_io.Reset(socket);
+ _io = &_socket_io;
+ _protocol_version = 0;
+ _local_enabled = false;
+ _phase.store(UNINITIALIZED, butil::memory_order_relaxed);
+ }
+
+ int phase(butil::memory_order order = butil::memory_order_acquire) const {
Review Comment:
Require the caller to explicitly pass the `memory_order` instead of using a
default argument.
##########
src/brpc/handshake/handshake_adapter.cpp:
##########
@@ -0,0 +1,55 @@
+// 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/handshake/handshake_adapter.h"
+
+#include "brpc/socket.h"
+
+namespace brpc {
+namespace handshake {
+
+// InputMessenger may call this entry repeatedly while bytes arrive. Keep all
+// connection state in HandshakeSession and Socket rather than in the adapter,
+// so a single stateless adapter can serve every connection. The parsing
+// context retains that selected adapter between the server hello and the peer
+// ACK, whose frame has no protocol magic of its own.
+ParseResult StandardHandshakeAdapter::ExecuteServerHandshake(
+ butil::IOBuf* source, Socket* socket) {
+ const StepResult result = RunServerStep(source, socket);
+ if (result == STEP_NEED_MORE) {
+ if (GetSession(socket)->phase() == ACK_WAIT &&
Review Comment:
`StandardHandshakeAdapter::ExecuteServerHandshake()` retains the selected
adapter only when the session reaches `ACK_WAIT`. UBSHM first enters
`EXTENSION_WAIT` after sending the server Hello.
The client sends its extension after receiving that Hello, so the extension
normally arrives in a subsequent read. Without a retained parsing context,
`ProcessUpgradeReadable()` attempts magic-based dispatch again. The extension
has no magic, so it returns `TRY_OTHERS` instead of continuing the handshake.
##########
src/brpc/handshake/handshake_adapter.cpp:
##########
@@ -0,0 +1,55 @@
+// 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/handshake/handshake_adapter.h"
+
+#include "brpc/socket.h"
+
+namespace brpc {
+namespace handshake {
+
+// InputMessenger may call this entry repeatedly while bytes arrive. Keep all
+// connection state in HandshakeSession and Socket rather than in the adapter,
+// so a single stateless adapter can serve every connection. The parsing
+// context retains that selected adapter between the server hello and the peer
+// ACK, whose frame has no protocol magic of its own.
+ParseResult StandardHandshakeAdapter::ExecuteServerHandshake(
+ butil::IOBuf* source, Socket* socket) {
+ const StepResult result = RunServerStep(source, socket);
+ if (result == STEP_NEED_MORE) {
+ if (GetSession(socket)->phase() == ACK_WAIT &&
+ socket->parsing_context() == NULL) {
+ ServerHandshakeContext* context =
+ ServerHandshakeContext::Create(this);
+ if (context == NULL) {
+ GetSession(socket)->MarkFailed();
+ return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
+ }
+ socket->reset_parsing_context(context);
+ }
+ return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
+ }
+
+ socket->reset_parsing_context(NULL);
+ if (result == STEP_ERROR) {
+ return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
+ }
+ return MakeParseError(PARSE_ERROR_TRY_OTHERS);
Review Comment:
After a successful handshake, the adapter clears its parsing context and
returns `TRY_OTHERS`, allowing ordinary RPC parsers to take over.
`OnNewMessagesAfterUpgrade()` checks `ESTABLISHED` only before entering
`InputMessenger::OnNewMessages()`. If the handshake completes inside that call,
the same TCP read loop can read subsequent application bytes and dispatch them
to an RPC parser before the outer state check runs again.
This violates the intended behavior that TCP remains a control connection
after upgrading.
##########
test/brpc_rdma_unittest.cpp:
##########
@@ -376,144 +358,192 @@ TEST_F(RdmaTest,
stale_cq_callback_does_not_poll_new_generation) {
TEST_F(RdmaTest, client_close_before_hello_send) {
StartServer();
- butil::fd_guard sockfd;
- ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd));
- Socket* s = WaitForServerSocket();
- ASSERT_TRUE(s != nullptr);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT,
RdmaTransportOf(s)->_rdma_ep->_state);
- sockfd.reset(-1);
- ASSERT_TRUE(WaitForServerSocketGone());
+ sockaddr_in addr;
+ bzero((char*)&addr, sizeof(addr));
+ addr.sin_family = AF_INET;
+ addr.sin_port = htons(PORT);
+
+ butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0));
+ ASSERT_TRUE(sockfd >= 0);
+ ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
+ usleep(100000); // wait for server to handle the msg
+ Socket* s = GetSocketFromServer(0);
+ ASSERT_EQ(handshake::UNINITIALIZED,
+ AdapterTransport::Get(s)->handshake_phase());
+ close(sockfd);
+ usleep(100000); // wait for server to handle the msg
+ ASSERT_EQ(nullptr, GetSocketFromServer(0));
StopServer();
}
TEST_F(RdmaTest, client_hello_msg_invalid_magic_str) {
StartServer();
- butil::fd_guard sockfd;
- ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd));
- Socket* s = WaitForServerSocket();
- ASSERT_TRUE(s != nullptr);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT,
RdmaTransportOf(s)->_rdma_ep->_state);
+ sockaddr_in addr;
+ bzero((char*)&addr, sizeof(addr));
+ addr.sin_family = AF_INET;
+ addr.sin_port = htons(PORT);
+
+ butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0));
+ ASSERT_TRUE(sockfd >= 0);
+ ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
+ usleep(100000); // wait for server to handle the msg
+ Socket* s = GetSocketFromServer(0);
+ ASSERT_EQ(handshake::UNINITIALIZED,
+ AdapterTransport::Get(s)->handshake_phase());
uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
memcpy(data, "PRPC", 4); // send as normal baidu_std protocol
- ASSERT_TRUE(WriteAll(sockfd, data, 4));
- // Wait for the bytes to show up in the fd stream (baidu_std wants 12B of
- // header, so they stay buffered). Waiting on the state instead would prove
- // nothing: it is already UNINIT before the server has read anything.
- ASSERT_TRUE(WaitForFdReadBuf(s, 4));
- // 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,
RdmaTransportOf(s)->_rdma_ep->_state);
+ ASSERT_EQ(4, write(sockfd, data, 4));
+ usleep(100000); // wait for server to handle the msg
+ // A non-RDMA magic makes the transport-handshake parser 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(handshake::UNINITIALIZED,
+ AdapterTransport::Get(s)->handshake_phase());
Review Comment:
No need to modify.
##########
src/brpc/policy/transport_handshake_protocol.cpp:
##########
@@ -0,0 +1,37 @@
+// 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/transport_handshake_protocol.h"
+
+#include "butil/logging.h"
+#include "brpc/adapter_transport.h"
+
+namespace brpc {
+namespace policy {
+
+ParseResult ParseTransportHandshake(butil::IOBuf* source, Socket* socket,
+ bool /*read_eof*/, const void* /*arg*/) {
+ return AdapterTransport::Get(socket)->ProcessUpgradeReadable(source);
Review Comment:
`ParseTransportHandshake()` unconditionally calls
`AdapterTransport::Get(socket)`, which performs a `static_cast` from the
socket's transport.
`TransportFactory` still installs `UrmaTransport` for URMA sockets. When
URMA falls back to TCP, its input reaches `InputMessenger` and the globally
registered handshake parser can participate in protocol detection. This casts
an unrelated object to `AdapterTransport`.
Even when the input does not match a handshake magic,
`ProcessUpgradeReadable()` accesses handshake and connection-completion fields.
These accesses produce undefined behavior and may crash.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
