Changed RDMA testing server not to use the lower level Rdma buffers

if it needs to queue messages to echo, but to suffer the double copy
overhead of using its own buffers

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

1 files changed, 35 insertions, 11 deletions

diff --git a/qpid/cpp/src/qpid/sys/rdma/RdmaServer.cpp b/qpid/cpp/src/qpid/sys/rdma/RdmaServer.cpp
index 97715326d5..564fd62730 100644
--- a/qpid/cpp/src/qpid/sys/rdma/RdmaServer.cpp
+++ b/qpid/cpp/src/qpid/sys/rdma/RdmaServer.cpp
@@ -44,10 +44,28 @@ using qpid::sys::Dispatcher;
 namespace qpid {
 namespace tests {
 
+struct Buffer {
+    char* bytes() const {return bytes_;}
+    int32_t byteCount() const {return size;}
+
+    Buffer(const int32_t s):
+        bytes_(new char[s]),
+        size(s)
+    {
+    }
+
+    ~Buffer() {
+        delete [] bytes_;
+    }
+private:
+    char* bytes_;
+    int32_t size;
+};
+
 struct ConRec {
     Rdma::Connection::intrusive_ptr connection;
     Rdma::AsynchIO* data;
-    queue<Rdma::Buffer*> queuedWrites;
+    queue<Buffer*> queuedWrites;
 
     ConRec(Rdma::Connection::intrusive_ptr c) :
         connection(c)
@@ -60,30 +78,36 @@ void dataError(Rdma::AsynchIO&) {
 
 void idle(ConRec* cr, Rdma::AsynchIO& a) {
     // Need to make sure full is not called as it would reorder messages
-    while (!cr->queuedWrites.empty() && a.writable()) {
-        Rdma::Buffer* buf = cr->queuedWrites.front();
+    while (!cr->queuedWrites.empty() && a.writable() && a.bufferAvailable()) {
+        Buffer* buf = cr->queuedWrites.front();
         cr->queuedWrites.pop();
-        a.queueWrite(buf);
+        Rdma::Buffer* rbuf = a.getBuffer();
+        std::copy(buf->bytes(), buf->bytes()+buf->byteCount(), rbuf->bytes());
+        rbuf->dataCount(buf->byteCount());
+        delete buf;
+        a.queueWrite(rbuf);
     }
 }
 
 void data(ConRec* cr, Rdma::AsynchIO& a, Rdma::Buffer* b) {
     // Echo data back
-    Rdma::Buffer* buf = a.getBuffer();
-    std::copy(b->bytes(), b->bytes()+b->dataCount(), buf->bytes());
-    buf->dataCount(b->dataCount());
-    if (cr->queuedWrites.empty()) {
-        // If can't write then full will be called and push buffer on back of queue
+    if (cr->queuedWrites.empty() && a.writable() && a.bufferAvailable()) {
+        Rdma::Buffer* buf = a.getBuffer();
+        std::copy(b->bytes(), b->bytes()+b->dataCount(), buf->bytes());
+        buf->dataCount(b->dataCount());
         a.queueWrite(buf);
     } else {
+        Buffer* buf = new Buffer(b->dataCount());
+        std::copy(b->bytes(), b->bytes()+b->dataCount(), buf->bytes());
         cr->queuedWrites.push(buf);
         // Try to empty queue
         idle(cr, a);
     }
 }
 
-void full(ConRec* cr, Rdma::AsynchIO&, Rdma::Buffer* buf) {
-    cr->queuedWrites.push(buf);
+void full(ConRec*, Rdma::AsynchIO&, Rdma::Buffer*) {
+    // Shouldn't ever be called
+    cout << "!";
 }
 
 void drained(Rdma::AsynchIO&) {