synchronized with trunk except for ruby dir

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/qpid.rnr@893970 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 353 insertions, 0 deletions

diff --git a/cpp/src/qpid/sys/RdmaIOPlugin.cpp b/cpp/src/qpid/sys/RdmaIOPlugin.cpp
new file mode 100644
index 0000000000..bd19247124
--- /dev/null
+++ b/cpp/src/qpid/sys/RdmaIOPlugin.cpp
@@ -0,0 +1,353 @@
+/*
+ *
+ * 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 "qpid/sys/ProtocolFactory.h"
+
+#include "qpid/Plugin.h"
+#include "qpid/broker/Broker.h"
+#include "qpid/framing/AMQP_HighestVersion.h"
+#include "qpid/log/Statement.h"
+#include "qpid/sys/rdma/RdmaIO.h"
+#include "qpid/sys/OutputControl.h"
+
+#include <boost/bind.hpp>
+#include <boost/lexical_cast.hpp>
+#include <memory>
+
+#include <netdb.h>
+
+using std::auto_ptr;
+using std::string;
+using std::stringstream;
+
+namespace qpid {
+namespace sys {
+
+class RdmaIOHandler : public OutputControl {
+    Rdma::Connection::intrusive_ptr connection;
+    std::string identifier;
+    Rdma::AsynchIO* aio;
+    ConnectionCodec::Factory* factory;
+    ConnectionCodec* codec;
+    bool readError;
+
+    void write(const framing::ProtocolInitiation&);
+
+  public:
+    RdmaIOHandler(Rdma::Connection::intrusive_ptr& c, ConnectionCodec::Factory* f);
+    ~RdmaIOHandler();
+    void init(Rdma::AsynchIO* a);
+    void start(Poller::shared_ptr poller) {aio->start(poller);}
+
+    // Output side
+    void close();
+    void abort();
+    void activateOutput();
+    void giveReadCredit(int32_t credit);
+    void initProtocolOut();
+
+    // Input side
+    void readbuff(Rdma::AsynchIO& aio, Rdma::Buffer* buff);
+    void initProtocolIn(Rdma::Buffer* buff);
+
+    // Notifications
+    void full(Rdma::AsynchIO& aio);
+    void idle(Rdma::AsynchIO& aio);
+    void error(Rdma::AsynchIO& aio);
+};
+
+RdmaIOHandler::RdmaIOHandler(Rdma::Connection::intrusive_ptr& c, qpid::sys::ConnectionCodec::Factory* f) :
+    connection(c),
+    identifier(c->getPeerName()),
+    factory(f),
+    codec(0),
+    readError(false)
+{
+}
+
+void RdmaIOHandler::init(Rdma::AsynchIO* a) {
+    aio = a;
+}
+
+RdmaIOHandler::~RdmaIOHandler() {
+    if (codec)
+        codec->closed();
+    delete codec;
+
+    aio->deferDelete();
+}
+
+void RdmaIOHandler::write(const framing::ProtocolInitiation& data)
+{
+    QPID_LOG(debug, "Rdma: SENT [" << identifier << "] INIT(" << data << ")");
+    Rdma::Buffer* buff = aio->getBuffer();
+    framing::Buffer out(buff->bytes, buff->byteCount);
+    data.encode(out);
+    buff->dataCount = data.encodedSize();
+    aio->queueWrite(buff);
+}
+
+void RdmaIOHandler::close() {
+    aio->queueWriteClose();
+}
+
+// TODO: Dummy implementation, need to fill this in for heartbeat timeout to work
+void RdmaIOHandler::abort() {
+}
+
+void RdmaIOHandler::activateOutput() {
+    aio->notifyPendingWrite();
+}
+
+void RdmaIOHandler::idle(Rdma::AsynchIO&) {
+    // TODO: Shouldn't need this test as idle() should only ever be called when
+    // the connection is writable anyway
+    if ( !(aio->writable() && aio->bufferAvailable()) ) {
+        return;
+    }
+    if (codec == 0) return;
+    if (codec->canEncode()) {
+        Rdma::Buffer* buff = aio->getBuffer();
+        size_t encoded=codec->encode(buff->bytes, buff->byteCount);
+        buff->dataCount = encoded;
+        aio->queueWrite(buff);
+    }
+    if (codec->isClosed())
+        aio->queueWriteClose();
+}
+
+void RdmaIOHandler::initProtocolOut() {
+    // We mustn't have already started the conversation
+    // but we must be able to send
+    assert( codec == 0 );
+    assert( aio->writable() && aio->bufferAvailable() );
+    codec = factory->create(*this, identifier, 0);
+    write(framing::ProtocolInitiation(codec->getVersion()));
+}
+
+void RdmaIOHandler::error(Rdma::AsynchIO&) {
+    close();
+}
+
+void RdmaIOHandler::full(Rdma::AsynchIO&) {
+    QPID_LOG(debug, "Rdma: buffer full [" << identifier << "]");
+}
+
+// TODO: Dummy implementation of read throttling
+void RdmaIOHandler::giveReadCredit(int32_t) {
+}
+
+// The logic here is subtly different from TCP as RDMA is message oriented
+// so we define that an RDMA message is a frame - in this case there is no putting back
+// of any message remainder - there shouldn't be any. And what we read here can't be
+// smaller than a frame
+void RdmaIOHandler::readbuff(Rdma::AsynchIO&, Rdma::Buffer* buff) {
+    if (readError) {
+        return;
+    }
+    size_t decoded = 0;
+    try {
+        if (codec) {
+            decoded = codec->decode(buff->bytes+buff->dataStart, buff->dataCount);
+        }else{
+            // Need to start protocol processing
+            initProtocolIn(buff);
+        }
+    }catch(const std::exception& e){
+        QPID_LOG(error, e.what());
+        readError = true;
+        aio->queueWriteClose();
+    }
+}
+
+void RdmaIOHandler::initProtocolIn(Rdma::Buffer* buff) {
+    framing::Buffer in(buff->bytes+buff->dataStart, buff->dataCount);
+    framing::ProtocolInitiation protocolInit;
+    size_t decoded = 0;
+    if (protocolInit.decode(in)) {
+        decoded = in.getPosition();
+        QPID_LOG(debug, "Rdma: RECV [" << identifier << "] INIT(" << protocolInit << ")");
+
+        codec = factory->create(protocolInit.getVersion(), *this, identifier, 0);
+
+        // If we failed to create the codec then we don't understand the offered protocol version
+        if (!codec) {
+            // send valid version header & close connection.
+            write(framing::ProtocolInitiation(framing::highestProtocolVersion));
+            readError = true;
+            aio->queueWriteClose();
+        }
+    }
+}
+
+class RdmaIOProtocolFactory : public ProtocolFactory {
+    auto_ptr<Rdma::Listener> listener;
+    const uint16_t listeningPort;
+
+  public:
+    RdmaIOProtocolFactory(int16_t port, int backlog);
+    void accept(Poller::shared_ptr, ConnectionCodec::Factory*);
+    void connect(Poller::shared_ptr, const string& host, int16_t port, ConnectionCodec::Factory*, ConnectFailedCallback);
+
+    uint16_t getPort() const;
+    string getHost() const;
+
+  private:
+    bool request(Rdma::Connection::intrusive_ptr&, const Rdma::ConnectionParams&, ConnectionCodec::Factory*);
+    void established(Poller::shared_ptr, Rdma::Connection::intrusive_ptr&);
+    void connected(Poller::shared_ptr, Rdma::Connection::intrusive_ptr&, const Rdma::ConnectionParams&, ConnectionCodec::Factory*);
+    void connectionError(Rdma::Connection::intrusive_ptr&, Rdma::ErrorType);
+    void disconnected(Rdma::Connection::intrusive_ptr&);
+    void rejected(Rdma::Connection::intrusive_ptr&, const Rdma::ConnectionParams&, ConnectFailedCallback);
+};
+
+// Static instance to initialise plugin
+static class RdmaIOPlugin : public Plugin {
+    void earlyInitialize(Target&) {
+    }
+
+    void initialize(Target& target) {
+        // Check whether we actually have any rdma devices
+        if ( Rdma::deviceCount() == 0 ) {
+            QPID_LOG(info, "Rdma: Disabled: no rdma devices found");
+            return;
+        }
+
+        broker::Broker* broker = dynamic_cast<broker::Broker*>(&target);
+        // Only provide to a Broker
+        if (broker) {
+            const broker::Broker::Options& opts = broker->getOptions();
+            ProtocolFactory::shared_ptr protocol(new RdmaIOProtocolFactory(opts.port, opts.connectionBacklog));
+            QPID_LOG(notice, "Rdma: Listening on RDMA port " << protocol->getPort());
+            broker->registerProtocolFactory("rdma", protocol);
+        }
+    }
+} rdmaPlugin;
+
+RdmaIOProtocolFactory::RdmaIOProtocolFactory(int16_t port, int /*backlog*/) :
+    listeningPort(port)
+{}
+
+void RdmaIOProtocolFactory::established(Poller::shared_ptr poller, Rdma::Connection::intrusive_ptr& ci) {
+    RdmaIOHandler* async = ci->getContext<RdmaIOHandler>();
+    async->start(poller);
+}
+
+bool RdmaIOProtocolFactory::request(Rdma::Connection::intrusive_ptr& ci, const Rdma::ConnectionParams& cp,
+        ConnectionCodec::Factory* f) {
+    try {
+        RdmaIOHandler* async = new RdmaIOHandler(ci, f);
+        Rdma::AsynchIO* aio =
+            new Rdma::AsynchIO(ci->getQueuePair(),
+                cp.maxRecvBufferSize, cp.initialXmitCredit, Rdma::DEFAULT_WR_ENTRIES,
+                boost::bind(&RdmaIOHandler::readbuff, async, _1, _2),
+                boost::bind(&RdmaIOHandler::idle, async, _1),
+                0, // boost::bind(&RdmaIOHandler::full, async, _1),
+                boost::bind(&RdmaIOHandler::error, async, _1));
+        async->init(aio);
+
+        // Record aio so we can get it back from a connection
+        ci->addContext(async);
+        return true;
+    } catch (const Rdma::Exception& e) {
+        QPID_LOG(error, "Rdma: Cannot accept new connection (Rdma exception): " << e.what());
+    } catch (const std::exception& e) {
+        QPID_LOG(error, "Rdma: Cannot accept new connection (unknown exception): " << e.what());
+    }
+
+    // If we get here we caught an exception so reject connection
+    return false;
+}
+
+void RdmaIOProtocolFactory::connectionError(Rdma::Connection::intrusive_ptr&, Rdma::ErrorType) {
+}
+
+void RdmaIOProtocolFactory::disconnected(Rdma::Connection::intrusive_ptr& ci) {
+    // If we've got a connection already tear it down, otherwise ignore
+    RdmaIOHandler* async =  ci->getContext<RdmaIOHandler>();
+    if (async) {
+        async->close();
+    }
+    delete async;
+}
+
+uint16_t RdmaIOProtocolFactory::getPort() const {
+    return listeningPort; // Immutable no need for lock.
+}
+
+string RdmaIOProtocolFactory::getHost() const {
+    //return listener.getSockname();
+    return "";
+}
+
+void RdmaIOProtocolFactory::accept(Poller::shared_ptr poller, ConnectionCodec::Factory* fact) {
+    ::sockaddr_in sin;
+
+    sin.sin_family = AF_INET;
+    sin.sin_port = htons(listeningPort);
+    sin.sin_addr.s_addr = INADDR_ANY;
+
+    SocketAddress sa("",boost::lexical_cast<std::string>(listeningPort));
+    listener.reset(
+        new Rdma::Listener(sa,
+            Rdma::ConnectionParams(65536, Rdma::DEFAULT_WR_ENTRIES),
+            boost::bind(&RdmaIOProtocolFactory::established, this, poller, _1),
+            boost::bind(&RdmaIOProtocolFactory::connectionError, this, _1, _2),
+            boost::bind(&RdmaIOProtocolFactory::disconnected, this, _1),
+            boost::bind(&RdmaIOProtocolFactory::request, this, _1, _2, fact)));
+
+    listener->start(poller);
+}
+
+// Only used for outgoing connections (in federation)
+void RdmaIOProtocolFactory::rejected(Rdma::Connection::intrusive_ptr&, const Rdma::ConnectionParams&, ConnectFailedCallback failed) {
+    failed(-1, "Connection rejected");
+}
+
+// Do the same as connection request and established but mark a client too
+void RdmaIOProtocolFactory::connected(Poller::shared_ptr poller, Rdma::Connection::intrusive_ptr& ci, const Rdma::ConnectionParams& cp,
+        ConnectionCodec::Factory* f) {
+    (void) request(ci, cp, f);
+    established(poller, ci);
+    RdmaIOHandler* async =  ci->getContext<RdmaIOHandler>();
+    async->initProtocolOut();
+}
+
+void RdmaIOProtocolFactory::connect(
+    Poller::shared_ptr poller,
+    const std::string& host, int16_t port,
+    ConnectionCodec::Factory* f,
+    ConnectFailedCallback failed)
+{
+    SocketAddress sa(host, boost::lexical_cast<std::string>(port));
+    Rdma::Connector* c =
+        new Rdma::Connector(
+            sa,
+            Rdma::ConnectionParams(8000, Rdma::DEFAULT_WR_ENTRIES),
+            boost::bind(&RdmaIOProtocolFactory::connected, this, poller, _1, _2, f),
+            boost::bind(&RdmaIOProtocolFactory::connectionError, this, _1, _2),
+            boost::bind(&RdmaIOProtocolFactory::disconnected, this, _1),
+            boost::bind(&RdmaIOProtocolFactory::rejected, this, _1, _2, failed));
+
+    c->start(poller);
+}
+
+}} // namespace qpid::sys