chenBright commented on code in PR #3535:
URL: https://github.com/apache/brpc/pull/3535#discussion_r4489903011
##########
src/brpc/transport_factory.cpp:
##########
@@ -49,11 +50,11 @@ int TransportFactory::ContextInitOrDie(
std::unique_ptr<Transport> TransportFactory::CreateTransport(SocketMode mode) {
if (mode == SOCKET_MODE_TCP) {
- return std::unique_ptr<TcpTransport>(new TcpTransport());
+ return std::unique_ptr<Transport>(new AdapterTransport(mode));
Review Comment:
Please use `std::make_unique` instead.
##########
src/brpc/transport_handshake.cpp:
##########
@@ -0,0 +1,427 @@
+// 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/transport_handshake.h"
+
+#include <errno.h>
+#include <cstring>
+
+#include "butil/logging.h"
+#include "butil/object_pool.h"
+#include "butil/sys_byteorder.h"
+
+namespace brpc {
+namespace handshake {
+
+namespace {
+
+const size_t COMMON_ACK_SIZE = sizeof(uint32_t);
+const uint32_t COMMON_ACK_OK = 0x1;
+
+} // namespace
+
+StepResult HandshakeProtocol::BuildAck(bool enabled, std::string* payload) {
+ CHECK(payload != nullptr);
+ const uint32_t flags = butil::HostToNet32(enabled ? COMMON_ACK_OK : 0);
+ payload->assign(reinterpret_cast<const char*>(&flags), sizeof(flags));
+ return STEP_OK;
+}
+
+StepResult HandshakeProtocol::ParseAck(const std::string& payload,
+ bool* enabled) {
+ CHECK(enabled != nullptr);
+ if (payload.size() != COMMON_ACK_SIZE) {
+ errno = EPROTO;
+ return STEP_ERROR;
+ }
+ uint32_t flags = 0;
+ memcpy(&flags, payload.data(), sizeof(flags));
+ *enabled = (butil::NetToHost32(flags) & COMMON_ACK_OK) != 0;
+ return STEP_OK;
+}
+
+const FrameSpec& HandshakeProtocol::ExtensionFrameSpec() const {
+ static const FrameSpec empty_spec(
+ nullptr, 0, 0, 0, FrameSpec::FIXED);
+ return empty_spec;
+}
+
+ServerHandshakeContext* ServerHandshakeContext::Create(
+ HandshakeAdapter* adapter) {
+ ServerHandshakeContext* context =
+ butil::get_object<ServerHandshakeContext>();
+ if (context != nullptr) {
+ context->_adapter = adapter;
+ }
+ return context;
+}
+
+void ServerHandshakeContext::Destroy() {
+ _adapter = nullptr;
+ butil::return_object(this);
+}
+
+static StepResult FinishWithFailure(HandshakeSession* session,
+ HandshakeTransport* transport) {
+ const int saved_errno = errno != 0 ? errno : EPROTO;
+ transport->OnFailed();
+ session->MarkFailed();
+ errno = saved_errno;
+ return STEP_ERROR;
+}
+
+static StepResult FinishWithFallback(HandshakeSession* session,
+ HandshakeTransport* transport) {
+ session->PublishFallback([transport]() { transport->OnFallback(); });
+ return STEP_FALLBACK;
+}
+
+static StepResult ConvertFrameResult(FrameResult result) {
+ switch (result) {
+ case FRAME_OK: return STEP_OK;
+ case FRAME_NOT_MINE: return STEP_NOT_MINE;
+ case FRAME_NEED_MORE: return STEP_NEED_MORE;
+ case FRAME_IO_ERROR: return STEP_ERROR;
+ case FRAME_PROTOCOL_ERROR:
+ errno = EPROTO;
+ return STEP_ERROR;
+ }
+ errno = EPROTO;
+ return STEP_ERROR;
+}
+
+StepResult HandshakeSession::SendHello(HandshakeProtocol* protocol,
+ bool enabled) {
+ std::string payload;
+ const StepResult result = protocol->BuildHello(enabled, &payload);
+ if (result != STEP_OK) {
+ return result;
+ }
+ return ConvertFrameResult(
+ FrameCodec::WriteFrame(_io, protocol->HelloFrameSpec(), payload));
+}
+
+StepResult HandshakeSession::ReceiveHello(HandshakeProtocol* protocol,
+ butil::IOBuf* input,
+ bool push_back_on_not_mine,
+ bool* magic_matched) {
+ std::string payload;
+ const FrameResult frame_result = input != nullptr
+ ? FrameCodec::ParseBufferedFrame(
+ input, protocol->HelloFrameSpec(), &payload, magic_matched)
+ : FrameCodec::ReadFrame(
+ _io, protocol->HelloFrameSpec(), push_back_on_not_mine,
+ &payload);
+ const StepResult result = ConvertFrameResult(frame_result);
+ if (result != STEP_OK) {
+ return result;
+ }
+ set_protocol_version(protocol->ProtocolVersion());
+ return protocol->ParseHello(payload);
+}
+
+StepResult HandshakeSession::SendAck(HandshakeProtocol* protocol,
+ bool enabled) {
+ std::string payload;
+ const StepResult result = protocol->BuildAck(enabled, &payload);
+ if (result != STEP_OK) {
+ return result;
+ }
+ return ConvertFrameResult(
+ FrameCodec::WriteFrame(_io, protocol->AckFrameSpec(), payload));
+}
+
+StepResult HandshakeSession::ReceiveAck(HandshakeProtocol* protocol,
+ butil::IOBuf* input,
+ bool* enabled) {
+ std::string payload;
+ const FrameResult frame_result = input != nullptr
+ ? FrameCodec::ParseBufferedFrame(
+ input, protocol->AckFrameSpec(), &payload)
+ : FrameCodec::ReadFrame(
+ _io, protocol->AckFrameSpec(), false, &payload);
+ const StepResult result = ConvertFrameResult(frame_result);
+ if (result != STEP_OK) {
+ return result;
+ }
+ return protocol->ParseAck(payload, enabled);
+}
+
+StepResult HandshakeSession::SendExtension(HandshakeProtocol* protocol,
+ bool enabled) {
+ std::string payload;
+ const StepResult result = protocol->BuildExtension(enabled, &payload);
+ if (result != STEP_OK) {
+ return result;
+ }
+ return ConvertFrameResult(
+ FrameCodec::WriteFrame(
+ _io, protocol->ExtensionFrameSpec(), payload));
+}
+
+StepResult HandshakeSession::ReceiveExtension(HandshakeProtocol* protocol,
+ butil::IOBuf* input) {
+ std::string payload;
+ const FrameResult frame_result = input != nullptr
+ ? FrameCodec::ParseBufferedFrame(
+ input, protocol->ExtensionFrameSpec(), &payload)
+ : FrameCodec::ReadFrame(
+ _io, protocol->ExtensionFrameSpec(), false, &payload);
+ const StepResult result = ConvertFrameResult(frame_result);
+ return result == STEP_OK ? protocol->ParseExtension(payload) : result;
+}
+
+StepResult HandshakeSession::SelectAndReceiveHello(
+ const std::vector<HandshakeProtocol*>& protocols, butil::IOBuf* input,
+ bool push_back_on_not_mine, HandshakeProtocol** selected) {
+ CHECK(!protocols.empty());
+ CHECK(selected != nullptr);
+ if (input == nullptr) {
+ // A blocking byte stream cannot try a second codec after consuming
+ // bytes from the fd. Such protocols must select a single codec before
+ // entering the common session.
+ CHECK_EQ(1UL, protocols.size());
+ *selected = protocols.front();
+ return ReceiveHello(*selected, nullptr, push_back_on_not_mine);
+ }
+
+ bool need_more = false;
+ for (size_t i = 0; i < protocols.size(); ++i) {
+ bool magic_matched = false;
+ const StepResult result = ReceiveHello(
+ protocols[i], input, false, &magic_matched);
+ if (result == STEP_NOT_MINE) {
+ continue;
+ }
+ if (result == STEP_NEED_MORE) {
+ if (magic_matched) {
+ *selected = protocols[i];
+ set_protocol_version(protocols[i]->ProtocolVersion());
+ return STEP_NEED_MORE;
+ }
+ need_more = true;
+ continue;
+ }
+ *selected = protocols[i];
+ return result;
+ }
+ return need_more ? STEP_NEED_MORE : STEP_NOT_MINE;
+}
+
+StepResult HandshakeSession::RunClient(HandshakeProtocol* protocol,
+ HandshakeTransport* transport) {
+ CHECK(protocol != nullptr);
+ CHECK(transport != nullptr);
+ transport->OnProtocolSelected(protocol);
+ // A client handshake runs once on a potentially reused bthread. Do not
+ // let an errno left by earlier work override this handshake's result.
+ errno = 0;
+
+ SetPhase(PREPARING);
+ StepResult result = transport->PrepareResources();
+ if (result == STEP_FALLBACK) {
+ return FinishWithFallback(this, transport);
+ }
+ if (result != STEP_OK) {
+ return FinishWithFailure(this, transport);
+ }
+
+ SetPhase(HELLO_SEND);
+ if (SendHello(protocol, true) != STEP_OK) {
+ return FinishWithFailure(this, transport);
+ }
+
+ SetPhase(HELLO_WAIT);
+ result = ReceiveHello(protocol, nullptr, false);
+ if (result == STEP_NOT_MINE || result == STEP_NEED_MORE) {
+ errno = EPROTO;
+ }
+ if (result == STEP_ERROR || result == STEP_NOT_MINE ||
+ result == STEP_NEED_MORE) {
+ return FinishWithFailure(this, transport);
+ }
+ bool enabled = result == STEP_OK;
+
+ if (enabled && protocol->HasExtension()) {
+ SetPhase(EXTENSION_SEND);
+ if (SendExtension(protocol, true) != STEP_OK) {
+ return FinishWithFailure(this, transport);
+ }
+ SetPhase(EXTENSION_WAIT);
+ result = ReceiveExtension(protocol, nullptr);
+ if (result != STEP_OK && result != STEP_FALLBACK) {
+ return FinishWithFailure(this, transport);
+ }
+ enabled = result == STEP_OK;
+ }
+
+ if (enabled) {
+ SetPhase(NEGOTIATING);
+ result = transport->NegotiateResources();
+ if (result != STEP_OK && result != STEP_FALLBACK) {
+ return FinishWithFailure(this, transport);
+ }
+ enabled = result == STEP_OK;
+ }
+
+ SetPhase(ACK_SEND);
+ if (SendAck(protocol, enabled) != STEP_OK) {
+ return FinishWithFailure(this, transport);
+ }
+
+ if (enabled) {
+ transport->OnEstablished();
+ MarkEstablished();
+ return STEP_OK;
+ }
+ return FinishWithFallback(this, transport);
+}
+
+StepResult HandshakeSession::RunServer(
+ const std::vector<HandshakeProtocol*>& protocols, butil::IOBuf* input,
+ HandshakeTransport* transport, bool fallback_on_not_mine) {
+ CHECK(!protocols.empty());
+ CHECK(transport != nullptr);
+
+ // Once TCP fallback has been published, subsequent bytes are application
+ // protocol data and must bypass every upgrade codec without changing the
+ // terminal state.
+ if (phase(butil::memory_order_acquire) == FALLBACK_TCP) {
+ return STEP_NOT_MINE;
+ }
+
+ HandshakeProtocol* selected = nullptr;
+ if (phase(butil::memory_order_acquire) != ACK_WAIT &&
+ phase(butil::memory_order_acquire) != EXTENSION_WAIT) {
+ const int previous_phase = phase(butil::memory_order_acquire);
Review Comment:
Calling `phase()` once is sufficient.
##########
src/brpc/transport_handshake.h:
##########
@@ -0,0 +1,236 @@
+// 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(nullptr) {}
+ 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 let AdapterTransport read the handshake result uniformly.
+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 = nullptr)
+ : _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.store(0, butil::memory_order_relaxed);
+ _local_enabled = false;
+ _phase.store(UNINITIALIZED, butil::memory_order_relaxed);
+ }
+
+ int phase(butil::memory_order order) const {
+ return _phase.load(order);
+ }
+
+ void SetPhase(int phase) {
Review Comment:
`SetPhase` -> `set_phase`
--
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]
