chenBright commented on code in PR #3535:
URL: https://github.com/apache/brpc/pull/3535#discussion_r4369903866
##########
src/brpc/ubshm_transport.cpp:
##########
@@ -64,26 +72,58 @@ int UBShmTransport::Reset(int32_t expected_nref) {
}
std::shared_ptr<AppConnect> UBShmTransport::Connect() {
- if (_default_connect == nullptr) {
- return std::make_shared<ubring::UBConnect>();
- }
return _default_connect;
}
-int UBShmTransport::CutFromIOBuf(butil::IOBuf *buf) {
- if (_ub_ep && _ub_state != UB_OFF) {
- butil::IOBuf *data_arr[1] = {buf};
- return _ub_ep->CutFromIOBufList(data_arr, 1);
- } else {
- return _tcp_transport->CutFromIOBuf(buf);
+void UBShmTransport::SetHighSpeedAvailable(bool available) {
+ _ub_state = available ? UB_ON : UB_OFF;
+}
+
+int UBShmTransport::PrepareUpgradeResources(ubring::SHM *local_trx_shm,
+ const char *shm_name) {
+ return _ub_ep->AllocateClientResources(local_trx_shm, shm_name);
+}
+
+int UBShmTransport::NegotiateUpgradeResources(ubring::SHM *local_trx_shm,
+ const char *shm_name,
+ uint64_t remote_shm_len,
+ ubring::UbrDataFormat format) {
+ return _ub_ep->_ub_ring->UbrMapRemoteShm(
+ local_trx_shm, shm_name, remote_shm_len, format);
+}
+
+int UBShmTransport::PrepareServerUpgradeResources(ubring::SHM *remote_trx_shm,
+ ubring::SHM *local_trx_shm,
+ ubring::UbrDataFormat
format) {
+ return _ub_ep->AllocateServerResources(
+ remote_trx_shm, local_trx_shm, format);
+}
+
+void UBShmTransport::ActivateUpgrade() {
+ SetHighSpeedAvailable(true);
+}
+
+void UBShmTransport::DeactivateUpgrade() {
+ SetHighSpeedAvailable(false);
+ if (_ub_ep != nullptr) {
+ _ub_ep->Reset();
Review Comment:
When a UBSHM client connects to a TCP-only server, the peer never maps the
client’s shared memory. DeactivateUpgrade() synchronously calls Reset() →
UbrTrxClose(), which waits for a peer-close notification until the default
five-second timeout. This delays both TCP fallback and the
connection-completion callback, causing otherwise valid requests to time out.
##########
src/brpc/adapter_transport.cpp:
##########
@@ -0,0 +1,699 @@
+// 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/shm/shm_mgr.h"
+#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, ubring::UBShmHandshakeAdapter* protocol,
+ Socket* socket, int* connect_error)
+ : _transport(transport), _local_shm(local_shm),
+ _shm_name(shm_name), _protocol(protocol), _socket(socket),
+ _connect_error(connect_error) {}
+
+ handshake::StepResult PrepareResources() override {
+ if (_transport->PrepareUpgradeResources(
+ _local_shm, _shm_name.c_str()) != 0) {
+ return handshake::STEP_FALLBACK;
+ }
+ _protocol->ConfigureClientHello(*_local_shm);
+ return handshake::STEP_OK;
+ }
+
+ handshake::StepResult NegotiateResources() override {
+ return _transport->NegotiateUpgradeResources(
+ _local_shm, _shm_name.c_str(), _protocol->remote().len,
+ _protocol->local_format()) == 0
+ ? handshake::STEP_OK : handshake::STEP_FALLBACK;
+ }
+
+ void OnEstablished() override {
+ _transport->GetUBShmEp()->SetNegotiatedDataFormat(
+ _protocol->local_format());
+ _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;
+ ubring::UBShmHandshakeAdapter* _protocol;
+ 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 != nullptr);
+ return static_cast<AdapterTransport*>(socket->_transport.get());
+}
+
+const AdapterTransport* AdapterTransport::Get(const Socket* socket) {
+ CHECK(socket != nullptr);
+ return static_cast<const AdapterTransport*>(socket->_transport.get());
+}
+
+bool AdapterTransport::upgrade_capable(SocketMode mode) const {
+ if (_mode != mode || _high_speed_transport == nullptr) {
+ 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() != nullptr) {
+ handshake::ServerHandshakeContext* context =
+ static_cast<handshake::ServerHandshakeContext*>(
+ _socket->parsing_context());
+ CHECK(context->adapter() != nullptr);
+ result = context->adapter()->ExecuteServerHandshake(source, _socket);
+ } else if (!source->empty()) {
Review Comment:
The dispatcher does not guard the ESTABLISHED state. Subsequent input
recognized as another handshake can reach PrepareUpgradeResources() again. An
established RDMA endpoint already has a resource, so repeated allocation
triggers CHECK(_resource == nullptr).
--
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]
