Support RDMA handshake protocol

Author: chenBright

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,

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()) {

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 @@ class InputMessageClosure {
 class InputMessenger : public SocketUser {
 friend class Socket;
 friend class TcpTransport;
+friend class RdmaTransport;
 friend class rdma::RdmaEndpoint;
 public:
     explicit InputMessenger(size_t capacity = 128);

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 {

src/brpc/parse_result.h

@@ -19,6 +19,7 @@
 #ifndef BRPC_PARSE_RESULT_H
 #define BRPC_PARSE_RESULT_H
 
+#include <stddef.h>
 
 namespace brpc {
 

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

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 redis 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