Add The transport layer to support communication protocols of different device vendors.

Author: zchuango

example/rdma_performance/client.cpp

@@ -102,7 +102,7 @@ class PerformanceTest {
 
     int Init() {
         brpc::ChannelOptions options;
-        options.use_rdma = FLAGS_use_rdma;
+        options.socket_mode = FLAGS_use_rdma? RDMA : TCP;
         options.protocol = FLAGS_protocol;
         options.connection_type = FLAGS_connection_type;
         options.timeout_ms = FLAGS_rpc_timeout_ms;

example/rdma_performance/server.cpp

@@ -76,7 +76,7 @@ int main(int argc, char* argv[]) {
     g_last_time.store(0, butil::memory_order_relaxed);
 
     brpc::ServerOptions options;
-    options.use_rdma = FLAGS_use_rdma;
+    options.socket_mode = FLAGS_use_rdma? RDMA : TCP;
     if (server.Start(FLAGS_port, &options) != 0) {
         LOG(ERROR) << "Fail to start EchoServer";
         return -1;

src/brpc/acceptor.cpp

@@ -23,6 +23,7 @@
 #include "butil/time.h"                     // gettimeofday_us
 #include "brpc/rdma/rdma_endpoint.h"
 #include "brpc/acceptor.h"
+#include "brpc/transport_factory.h"
 
 
 namespace brpc {
@@ -40,7 +41,7 @@ Acceptor::Acceptor(bthread_keytable_pool_t* pool)
     , _empty_cond(&_map_mutex)
     , _force_ssl(false)
     , _ssl_ctx(NULL) 
-    , _use_rdma(false)
+    , socket_mode(TCP)
     , _bthread_tag(BTHREAD_TAG_DEFAULT) {
 }
 
@@ -282,18 +283,10 @@ void Acceptor::OnNewConnectionsUntilEAGAIN(Socket* acception) {
         options.fd = in_fd;
         butil::sockaddr2endpoint(&in_addr, in_len, &options.remote_side);
         options.user = acception->user();
+        options.need_on_edge_trigger = true;
         options.force_ssl = am->_force_ssl;
         options.initial_ssl_ctx = am->_ssl_ctx;
-#if BRPC_WITH_RDMA
-        if (am->_use_rdma) {
-            options.on_edge_triggered_events = rdma::RdmaEndpoint::OnNewDataFromTcp;
-        } else {
-#else
-        {
-#endif
-            options.on_edge_triggered_events = InputMessenger::OnNewMessages;
-        }
-        options.use_rdma = am->_use_rdma;
+        options.socket_mode = am->socket_mode;
         options.bthread_tag = am->_bthread_tag;
         if (Socket::Create(options, &socket_id) != 0) {
             LOG(ERROR) << "Fail to create Socket";

src/brpc/acceptor.h

@@ -22,6 +22,7 @@
 #include "butil/synchronization/condition_variable.h"
 #include "butil/containers/flat_map.h"
 #include "brpc/input_messenger.h"
+#include "brpc/common.h"
 
 
 namespace brpc {
@@ -110,8 +111,8 @@ friend class Server;
     bool _force_ssl;
     std::shared_ptr<SocketSSLContext> _ssl_ctx;
 
-    // Whether to use rdma or not
-    bool _use_rdma;
+    // Choose to use a certain socket: 0 TCP, 1 RDMA
+    Mode socket_mode;
 
     // Acceptor belongs to this tag
     bthread_tag_t _bthread_tag;

src/brpc/channel.cpp

@@ -37,6 +37,7 @@
 #include "brpc/details/usercode_backup_pool.h"       // TooManyUserCode
 #include "brpc/rdma/rdma_helper.h"
 #include "brpc/policy/esp_authenticator.h"
+#include "brpc/transport_factory.h"
 
 namespace brpc {
 
@@ -60,7 +61,7 @@ ChannelOptions::ChannelOptions()
     , connection_type(CONNECTION_TYPE_UNKNOWN)
     , succeed_without_server(true)
     , log_succeed_without_server(true)
-    , use_rdma(false)
+    , socket_mode(TCP)
     , auth(NULL)
     , backup_request_policy(NULL)
     , retry_policy(NULL)
@@ -120,7 +121,7 @@ static ChannelSignature ComputeChannelSignature(const ChannelOptions& opt) {
         } else {
             // All disabled ChannelSSLOptions are the same
         }
-        if (opt.use_rdma) {
+        if (opt.socket_mode == RDMA) {
             buf.append("|rdma");
         }
         butil::MurmurHash3_x64_128_Update(&mm_ctx, buf.data(), buf.size());
@@ -163,20 +164,6 @@ Channel::~Channel() {
     }
 }
 
-#if BRPC_WITH_RDMA
-static bool OptionsAvailableForRdma(const ChannelOptions* opt) {
-    if (opt->has_ssl_options()) {
-        LOG(WARNING) << "Cannot use SSL and RDMA at the same time";
-        return false;
-    }
-    if (!rdma::SupportedByRdma(opt->protocol.name())) {
-        LOG(WARNING) << "Cannot use " << opt->protocol.name()
-                     << " over RDMA";
-        return false;
-    }
-    return true;
-}
-#endif
 
 int Channel::InitChannelOptions(const ChannelOptions* options) {
     if (options) {  // Override default options if user provided one.
@@ -191,20 +178,8 @@ int Channel::InitChannelOptions(const ChannelOptions* options) {
         _options.hc_option.health_check_path = FLAGS_health_check_path;
         _options.hc_option.health_check_timeout_ms = FLAGS_health_check_timeout_ms;
     }
-    if (_options.use_rdma) {
-#if BRPC_WITH_RDMA
-        if (!OptionsAvailableForRdma(&_options)) {
-            return -1;
-        }
-        rdma::GlobalRdmaInitializeOrDie();
-        if (!rdma::InitPollingModeWithTag(bthread_self_tag())) {
-            return -1;
-        }
-#else
-        LOG(WARNING) << "Cannot use rdma since brpc does not compile with rdma";
-        return -1;
-#endif
-    }
+    auto ret = TransportFactory::ContextInitOrDie(options->socket_mode, false, &_options);
+    CHECK(ret == 0);
 
     _serialize_request = protocol->serialize_request;
     _pack_request = protocol->pack_request;
@@ -369,7 +344,7 @@ int Channel::InitSingle(const butil::EndPoint& server_addr_and_port,
         return -1;
     }
     if (SocketMapInsert(SocketMapKey(server_addr_and_port, sig),
-                        &_server_id, ssl_ctx, _options.use_rdma, _options.hc_option) != 0) {
+                        &_server_id, ssl_ctx, _options.socket_mode, _options.hc_option) != 0) {
         LOG(ERROR) << "Fail to insert into SocketMap";
         return -1;
     }
@@ -406,7 +381,7 @@ int Channel::Init(const char* ns_url,
     GetNamingServiceThreadOptions ns_opt;
     ns_opt.succeed_without_server = _options.succeed_without_server;
     ns_opt.log_succeed_without_server = _options.log_succeed_without_server;
-    ns_opt.use_rdma = _options.use_rdma;
+    ns_opt.socket_mode = _options.socket_mode;
     ns_opt.channel_signature = ComputeChannelSignature(_options);
     ns_opt.hc_option =  _options.hc_option;
     if (CreateSocketSSLContext(_options, &ns_opt.ssl_ctx) != 0) {

src/brpc/channel.h

@@ -37,6 +37,7 @@
 #include "brpc/backup_request_policy.h"
 #include "brpc/naming_service_filter.h"
 #include "brpc/health_check_option.h"
+#include "brpc/common.h"
 
 namespace brpc {
 
@@ -105,9 +106,9 @@ struct ChannelOptions {
     const ChannelSSLOptions& ssl_options() const { return *_ssl_options; }
     ChannelSSLOptions* mutable_ssl_options();
 
-    // Let this channel use rdma rather than tcp.
-    // Default: false
-    bool use_rdma;
+    // Let this channel Choose to use a certain socket: 0 TCP, 1 RDMA.
+    // Default: TCP
+    Mode socket_mode;
 
     // Turn on authentication for this channel if `auth' is not NULL.
     // Note `auth' will not be deleted by channel and must remain valid when

src/brpc/common.h

@@ -0,0 +1,24 @@
+// 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_COMMON_H
+#define BRPC_COMMON_H
+enum Mode {
+    TCP = 0,
+    RDMA = 1
+};
+#endif //BRPC_COMMON_H

src/brpc/details/naming_service_thread.cpp

@@ -126,7 +126,7 @@ void NamingServiceThread::Actions::ResetServers(
         //       to pick those Sockets with the right settings during OnAddedServers
         const SocketMapKey key(_added[i], _owner->_options.channel_signature);
         CHECK_EQ(0, SocketMapInsert(key, &tagged_id.id, _owner->_options.ssl_ctx,
-                                    _owner->_options.use_rdma, _owner->_options.hc_option));
+                                    _owner->_options.socket_mode, _owner->_options.hc_option));
         _added_sockets.push_back(tagged_id);
     }
 

src/brpc/details/naming_service_thread.h

@@ -27,6 +27,7 @@
 #include "brpc/naming_service.h"                // NamingService
 #include "brpc/naming_service_filter.h"         // NamingServiceFilter
 #include "brpc/socket_map.h"
+#include "brpc/common.h"
 
 namespace brpc {
 
@@ -45,11 +46,11 @@ struct GetNamingServiceThreadOptions {
     GetNamingServiceThreadOptions()
         : succeed_without_server(false)
         , log_succeed_without_server(true)
-        , use_rdma(false) {}
-    
+        , socket_mode(TCP) {}
+
     bool succeed_without_server;
     bool log_succeed_without_server;
-    bool use_rdma;
+    Mode socket_mode;
     HealthCheckOption hc_option;
     ChannelSignature channel_signature;
     std::shared_ptr<SocketSSLContext> ssl_ctx;

src/brpc/input_message_base.h

@@ -55,6 +55,7 @@ class InputMessageBase : public Destroyable {
 friend class InputMessenger;
 friend void* ProcessInputMessage(void*);
 friend class Stream;
+friend class Transport;
     int64_t _received_us;
     int64_t _base_real_us;
     SocketUniquePtr _socket;