Copilot commented on code in PR #3535: URL: https://github.com/apache/brpc/pull/3535#discussion_r4037175109
########## src/brpc/adapter_transport.cpp: ########## @@ -0,0 +1,612 @@ +// 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_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) {} + + 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; +}; + +} // 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); + rdma::ParsedHello remote{}; + handshake::ClientHandshakeCallbacks callbacks{}; + callbacks.codec = protocol->MakeCodec(&remote); + callbacks.transport.prepare_resources = [&]() { + if (transport->PrepareUpgradeResources() == 0) { + return handshake::STEP_OK; + } + errno = 0; + return handshake::STEP_FALLBACK; + }; + callbacks.transport.negotiate_resources = [&]() { + return transport->NegotiateUpgradeResources(remote, false) == 0 + ? handshake::STEP_OK : handshake::STEP_FALLBACK; + }; + callbacks.transport.set_high_speed_active = [transport]() { + transport->ActivateUpgrade(); + }; + callbacks.transport.set_tcp_active = [transport]() { + transport->DeactivateUpgrade(); + }; + callbacks.transport.on_failed = [&]() { + 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)); + }; + const handshake::StepResult result = adapter->_handshake.RunClient(callbacks); + 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::HelloMessage remote{}; + ubring::UBShmHandshakeAdapter wire; + handshake::ClientHandshakeCallbacks callbacks{}; + callbacks.codec = wire.MakeCodec(); + callbacks.codec.build_hello = [&](bool enabled, std::string* payload) { + CHECK(enabled); + return wire.BuildHello(true, local_shm_len, shm_name_str.c_str(), + payload); + }; + callbacks.codec.parse_hello = [&](const std::string& payload) { + return wire.ParseHello(payload, &remote); + }; + callbacks.transport.prepare_resources = [&]() { + return transport->PrepareUpgradeResources( + &local_trx_shm, shm_name_str.c_str()) == 0 + ? handshake::STEP_OK : handshake::STEP_FALLBACK; + }; + callbacks.transport.negotiate_resources = [&]() { + return transport->NegotiateUpgradeResources( + &local_trx_shm, shm_name_str.c_str()) == 0 + ? handshake::STEP_OK : handshake::STEP_FALLBACK; + }; + callbacks.transport.set_high_speed_active = [transport]() { + transport->ActivateUpgrade(); + }; + callbacks.transport.set_tcp_active = [transport]() { + transport->DeactivateUpgrade(); + }; + callbacks.transport.on_failed = [&]() { + 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)); + }; + const handshake::StepResult result = adapter->_handshake.RunClient(callbacks); + if (result == handshake::STEP_OK) { + transport->FinishUpgrade(); + } + 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 + + socket->SetFailed(EPROTO, "Unsupported client transport handshake"); + adapter->CompleteConnection(handshake::FAILED); + task->done(EPROTO, task->data); + return NULL; +} + +void AdapterTransport::Init(Socket* socket, const SocketOptions& options) { + CHECK_EQ(_mode, options.socket_mode); + _socket = socket; + _default_connect = options.app_connect; + _on_edge_trigger = options.on_edge_triggered_events; Review Comment: `InputMessenger::Create` supplies `options.on_edge_triggered_events = InputMessenger::OnNewMessages` for client sockets. This leaves that callback active for RDMA/UBSHM clients, while `ProcessClientHandshake` concurrently reads the same fd in its background bthread, so the input messenger can consume handshake bytes first. Route client-side high-speed control sockets through `OnNewDataFromTcp` (or otherwise serialize the readers) before starting the handshake. This issue also appears in the following locations of the same file: - line 428 - line 428 ########## src/brpc/handshake/ubshm_handshake.cpp: ########## @@ -0,0 +1,413 @@ +// 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/ubshm_handshake.h" + +#include <errno.h> +#include <cstdio> + +#include "butil/raw_pack.h" +#include "butil/sys_byteorder.h" +#include "brpc/adapter_transport.h" +#include "brpc/socket.h" + +#if BRPC_WITH_UBRING + +#include <array> +#include <cstring> + +#include "butil/logging.h" +#include "brpc/reloadable_flags.h" +#include "brpc/ubshm/common/common.h" +#include "brpc/ubshm/ub_endpoint.h" +#include "brpc/ubshm/ub_helper.h" +#include "brpc/ubshm/ubr_trx.h" +#include "brpc/ubshm_transport.h" + +#endif + +namespace brpc { +namespace handshake { +namespace ubshm_wire { + +static const char* const MAGIC = "UB"; +static const size_t MAGIC_LEN = 2; +static const size_t HELLO_LEN = 64; +static const size_t ACK_LEN = 4; +#if BRPC_WITH_UBRING +static const uint16_t HELLO_VERSION = 2; +static const uint16_t IMPL_VERSION = 1; +#endif // BRPC_WITH_UBRING +static const uint32_t ACK_OK = 0x1; + +static const FrameSpec& HelloFrameSpec() { + static const FrameSpec spec( + MAGIC, MAGIC_LEN, HELLO_LEN, HELLO_LEN, FrameSpec::FIXED); + return spec; +} + +static const FrameSpec& AckFrameSpec() { + static const FrameSpec spec( + NULL, 0, ACK_LEN, ACK_LEN, FrameSpec::FIXED); + return spec; +} + +} // namespace ubshm_wire +} // namespace handshake +} // namespace brpc + +#if BRPC_WITH_UBRING + +namespace brpc { +namespace ubring { + +DEFINE_int32(data_queue_size, 4, "data queue size for UB"); +DEFINE_bool(ub_trace_verbose, false, "Print log message verbosely"); +BRPC_VALIDATE_GFLAG(ub_trace_verbose, brpc::PassValidate); + +void HelloMessage::Serialize(void* data) const { + char* current_pos = static_cast<char*>(data); + const uint16_t net_msg_len = butil::HostToNet16(msg_len); + memcpy(current_pos, &net_msg_len, sizeof(net_msg_len)); + current_pos += sizeof(net_msg_len); + const uint16_t net_hello_ver = butil::HostToNet16(hello_ver); + memcpy(current_pos, &net_hello_ver, sizeof(net_hello_ver)); + current_pos += sizeof(net_hello_ver); + const uint16_t net_impl_ver = butil::HostToNet16(impl_ver); + memcpy(current_pos, &net_impl_ver, sizeof(net_impl_ver)); + current_pos += sizeof(net_impl_ver); + const uint64_t net_len = butil::HostToNet64(len); + memcpy(current_pos, &net_len, sizeof(net_len)); + current_pos += sizeof(net_len); + memcpy(current_pos, shm_name, SHM_MAX_NAME_BUFF_LEN); +} + +void HelloMessage::Deserialize(const void* data) { + const char* current_pos = static_cast<const char*>(data); + uint16_t net_msg_len; + memcpy(&net_msg_len, current_pos, sizeof(net_msg_len)); + msg_len = butil::NetToHost16(net_msg_len); + current_pos += sizeof(net_msg_len); + uint16_t net_hello_ver; + memcpy(&net_hello_ver, current_pos, sizeof(net_hello_ver)); + hello_ver = butil::NetToHost16(net_hello_ver); + current_pos += sizeof(net_hello_ver); + uint16_t net_impl_ver; + memcpy(&net_impl_ver, current_pos, sizeof(net_impl_ver)); + impl_ver = butil::NetToHost16(net_impl_ver); + current_pos += sizeof(net_impl_ver); + uint64_t net_len; + memcpy(&net_len, current_pos, sizeof(net_len)); + len = butil::NetToHost64(net_len); + current_pos += sizeof(net_len); + memcpy(shm_name, current_pos, SHM_MAX_NAME_BUFF_LEN); +} + +std::string HelloMessage::toString() const { + constexpr size_t MAX_LEN = + 16 + 6 + 16 + 6 + 16 + 6 + 20 + 6 + SHM_MAX_NAME_BUFF_LEN + 32; + std::array<char, MAX_LEN> buf; + const int n = snprintf( + buf.data(), buf.size(), + "msg_len=%u, hello_ver=%u, impl_ver=%u, len=%lu, shm_name=%.*s", + msg_len, hello_ver, impl_ver, + static_cast<unsigned long>(len), + static_cast<int>(SHM_MAX_NAME_BUFF_LEN), shm_name); + return std::string(buf.data(), static_cast<size_t>(n)); +} + +handshake::HandshakeCodec UBShmHandshakeAdapter::MakeCodec() const { + handshake::HandshakeCodec codec{}; + codec.protocol_version = 2; + codec.hello_frame = handshake::ubshm_wire::HelloFrameSpec(); + codec.ack_frame = handshake::ubshm_wire::AckFrameSpec(); + codec.build_ack = [](bool enabled, std::string* payload) { + const uint32_t flags_be = butil::HostToNet32( + enabled ? handshake::ubshm_wire::ACK_OK : 0); + payload->assign(reinterpret_cast<const char*>(&flags_be), + sizeof(flags_be)); + return handshake::STEP_OK; + }; + codec.parse_ack = [](const std::string& payload, bool* enabled) { + if (payload.size() != handshake::ubshm_wire::ACK_LEN) { + errno = EPROTO; + return handshake::STEP_ERROR; + } + uint32_t flags_be = 0; + memcpy(&flags_be, payload.data(), sizeof(flags_be)); + *enabled = (butil::NetToHost32(flags_be) & + handshake::ubshm_wire::ACK_OK) != 0; + return handshake::STEP_OK; + }; + return codec; +} + +handshake::StepResult UBShmHandshakeAdapter::BuildHello( + bool enabled, uint64_t len, const char* shm_name, + std::string* payload) const { + HelloMessage message{}; + message.msg_len = static_cast<uint16_t>( + handshake::ubshm_wire::HELLO_LEN); + if (enabled) { + message.hello_ver = handshake::ubshm_wire::HELLO_VERSION; + message.impl_ver = handshake::ubshm_wire::IMPL_VERSION; + message.len = len; + if (shm_name == NULL) { + errno = EINVAL; + return handshake::STEP_ERROR; + } + const size_t shm_name_len = + strnlen(shm_name, SHM_MAX_NAME_LEN); + memcpy(message.shm_name, shm_name, shm_name_len); + } + payload->assign( + handshake::ubshm_wire::HELLO_LEN - + handshake::ubshm_wire::MAGIC_LEN, + '\0'); + message.Serialize(&(*payload)[0]); + return handshake::STEP_OK; +} + +handshake::StepResult UBShmHandshakeAdapter::ParseHello( + const std::string& payload, HelloMessage* message) const { + if (payload.size() != handshake::ubshm_wire::HELLO_LEN - + handshake::ubshm_wire::MAGIC_LEN) { + errno = EPROTO; + return handshake::STEP_ERROR; + } + message->Deserialize(payload.data()); + if (message->msg_len < handshake::ubshm_wire::HELLO_LEN) { + errno = EPROTO; + return handshake::STEP_ERROR; + } + return NegotiationValid(*message) ? + handshake::STEP_OK : handshake::STEP_FALLBACK; +} + +bool UBShmHandshakeAdapter::NegotiationValid( + const HelloMessage& message) const { + return message.hello_ver == handshake::ubshm_wire::HELLO_VERSION && + message.impl_ver == handshake::ubshm_wire::IMPL_VERSION; +} + +} // namespace ubring +} // namespace brpc + +#endif // BRPC_WITH_UBRING + +namespace brpc { +namespace handshake { + +class UBShmServerHandshakeAdapter : public StandardHandshakeAdapter { +public: + UBShmServerHandshakeAdapter() = default; + +protected: + StepResult RunServerStep( + butil::IOBuf* source, Socket* socket) override; + HandshakeSession* GetSession(Socket* socket) const override; + +private: + StepResult RunFallbackServerHandshake( + butil::IOBuf* source, Socket* socket); +#if BRPC_WITH_UBRING + StepResult RunUBShmServerHandshake( + butil::IOBuf* source, Socket* socket); +#endif + + DISALLOW_COPY_AND_ASSIGN(UBShmServerHandshakeAdapter); +}; + + +static HandshakeCodec MakeUBShmFallbackCodec() { + HandshakeCodec codec{}; + codec.protocol_version = 2; + codec.hello_frame = ubshm_wire::HelloFrameSpec(); + codec.ack_frame = ubshm_wire::AckFrameSpec(); + codec.parse_hello = [](const std::string&) { + return STEP_FALLBACK; + }; + codec.build_hello = [](bool enabled, std::string* payload) { + if (enabled) { + errno = EPROTO; + return STEP_ERROR; + } + payload->assign( + ubshm_wire::HELLO_LEN - ubshm_wire::MAGIC_LEN, '\0'); + butil::RawPacker(&(*payload)[0]) + .pack16(static_cast<uint16_t>(ubshm_wire::HELLO_LEN)); + return STEP_OK; + }; + codec.build_ack = [](bool enabled, std::string* payload) { + const uint32_t flags_be = butil::HostToNet32( + enabled ? ubshm_wire::ACK_OK : 0); + payload->assign(reinterpret_cast<const char*>(&flags_be), + sizeof(flags_be)); + return STEP_OK; + }; + codec.parse_ack = [](const std::string& payload, bool* enabled) { + if (payload.size() != ubshm_wire::ACK_LEN) { + errno = EPROTO; + return STEP_ERROR; + } + *enabled = false; + return STEP_OK; + }; + return codec; +} + +HandshakeAdapter* GetUBShmServerHandshakeAdapter() { + static UBShmServerHandshakeAdapter adapter; + return &adapter; +} + +HandshakeSession* UBShmServerHandshakeAdapter::GetSession( + Socket* socket) const { + return AdapterTransport::Get(socket)->handshake_session(); +} + +StepResult UBShmServerHandshakeAdapter::RunFallbackServerHandshake( + butil::IOBuf* source, Socket* socket) { + IOBufHandshakeInput input(source); + ServerHandshakeCallbacks callbacks{}; + callbacks.fallback_on_not_mine = false; + callbacks.codecs.push_back(MakeUBShmFallbackCodec()); + callbacks.input = &input; + callbacks.transport.prepare_resources = []() { return STEP_OK; }; + callbacks.transport.negotiate_resources = []() { return STEP_OK; }; + callbacks.transport.set_high_speed_active = []() {}; + callbacks.transport.set_tcp_active = []() {}; + callbacks.transport.on_failed = []() {}; + return GetSession(socket)->RunServer(callbacks); +} + +#if BRPC_WITH_UBRING +StepResult UBShmServerHandshakeAdapter::RunUBShmServerHandshake( + butil::IOBuf* source, Socket* socket) { + UBShmTransport* transport = UBShmTransport::Get(socket); + CHECK(transport->GetUBShmEp() != NULL); + + ubring::HelloMessage remote{}; + ubring::UBShmHandshakeAdapter wire; + IOBufHandshakeInput input(source); + ServerHandshakeCallbacks callbacks{}; + callbacks.fallback_on_not_mine = false; + callbacks.input = &input; + HandshakeCodec codec = wire.MakeCodec(); + codec.parse_hello = [&](const std::string& payload) { + const StepResult result = wire.ParseHello(payload, &remote); + if (result == STEP_OK || result == STEP_FALLBACK) { + LOG_IF(INFO, ubring::FLAGS_ub_trace_verbose) + << "server receive handshake message : " + << remote.toString(); + } + if (result == STEP_FALLBACK) { + transport->DeactivateUpgrade(); + } + return result; + }; + codec.build_hello = [&](bool enabled, std::string* payload) { + const uint64_t len = enabled + ? static_cast<uint64_t>(ubring::FLAGS_data_queue_size) * + MB_TO_BYTE + : 0; + return wire.BuildHello( + enabled, len, enabled ? remote.shm_name : NULL, payload); + }; + callbacks.codecs.push_back(codec); + callbacks.transport.prepare_resources = [&]() { + if (!ubring::IsUBAvailable()) { + transport->DeactivateUpgrade(); + return STEP_FALLBACK; + } + ubring::SHM remote_trx_shm = { + NULL, remote.len, 0, {0}, + static_cast<uint32_t>(socket->fd())}; + strncpy(remote_trx_shm.name, remote.shm_name, + SHM_MAX_NAME_BUFF_LEN); Review Comment: A peer can fill all 48 bytes of `remote.shm_name` with non-NUL data. `strncpy(..., SHM_MAX_NAME_BUFF_LEN)` then leaves `remote_trx_shm.name` unterminated, and `ShmRemoteMalloc` subsequently calls `strlen` on it, reading past the fixed-size array. Reject a name without a terminator before resource allocation and copy it with an explicit bounded representation. -- 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]
