Use structures with much less dynamic allocation to hold rdma buffers

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

2 files changed, 22 insertions, 18 deletions

diff --git a/cpp/src/qpid/sys/rdma/rdma_wrap.cpp b/cpp/src/qpid/sys/rdma/rdma_wrap.cpp
index c286782c96..4883f0ce14 100644
--- a/cpp/src/qpid/sys/rdma/rdma_wrap.cpp
+++ b/cpp/src/qpid/sys/rdma/rdma_wrap.cpp
@@ -153,7 +153,7 @@ namespace Rdma {
         // Deallocate recv buffer memory
         if (smr) delete [] static_cast<char*>(smr->addr);
 
-        // The buffers ptr_deque automatically deletes all the buffers we've allocated
+        // The buffers vectors automatically deletes all the buffers we've allocated
     }
 
     // Create buffers to use for writing
@@ -167,29 +167,32 @@ namespace Rdma {
         // Allocate memory block for all receive buffers
         char* mem = new char [sendBufferCount * bufferSize];
         smr = regMr(pd.get(), mem, sendBufferCount * bufferSize, ::IBV_ACCESS_LOCAL_WRITE);
+        sendBuffers.reserve(sendBufferCount);
+        freeBuffers.reserve(sendBufferCount);
         for (int i = 0; i<sendBufferCount; ++i) {
             // Allocate xmit buffer
-            Buffer* b = new Buffer(smr->lkey, &mem[i*bufferSize], bufferSize);
-            buffers.push_front(b);
-            bufferQueue.push_back(b);
+            sendBuffers.push_back(Buffer(smr->lkey, &mem[i*bufferSize], bufferSize));
+            freeBuffers.push_back(i);
         }
     }
 
     Buffer* QueuePair::getBuffer() {
-        qpid::sys::ScopedLock<qpid::sys::Mutex> l(bufferQueueLock);
-        assert(!bufferQueue.empty());
-        Buffer* b = bufferQueue.back();
-        bufferQueue.pop_back();
+        qpid::sys::ScopedLock<qpid::sys::Mutex> l(bufferLock);
+        assert(!freeBuffers.empty());
+        Buffer* b = &sendBuffers[freeBuffers.back()];
+        freeBuffers.pop_back();
         return b;
     }
 
     void QueuePair::returnBuffer(Buffer* b) {
-        qpid::sys::ScopedLock<qpid::sys::Mutex> l(bufferQueueLock);
-        bufferQueue.push_back(b);
+        qpid::sys::ScopedLock<qpid::sys::Mutex> l(bufferLock);
+        int i = b - &sendBuffers[0];
+        assert(i >= 0 && i < int(sendBuffers.size()));
+        freeBuffers.push_back(i);
     }
 
     bool QueuePair::bufferAvailable() const {
-        return !bufferQueue.empty();
+        return !freeBuffers.empty();
     }
 
     void QueuePair::allocateRecvBuffers(int recvBufferCount, int bufferSize)
@@ -202,11 +205,11 @@ namespace Rdma {
         // Allocate memory block for all receive buffers
         char* mem = new char [recvBufferCount * bufferSize];
         rmr = regMr(pd.get(), mem, recvBufferCount * bufferSize, ::IBV_ACCESS_LOCAL_WRITE);
+        recvBuffers.reserve(recvBufferCount);
         for (int i = 0; i<recvBufferCount; ++i) {
             // Allocate recv buffer
-            Buffer* b = new Buffer(rmr->lkey, &mem[i*bufferSize], bufferSize);
-            buffers.push_front(b);
-            postRecv(b);
+            recvBuffers.push_back(Buffer(rmr->lkey, &mem[i*bufferSize], bufferSize));
+            postRecv(&recvBuffers[i]);
         }
     }
 
diff --git a/cpp/src/qpid/sys/rdma/rdma_wrap.h b/cpp/src/qpid/sys/rdma/rdma_wrap.h
index f951dcb0af..51cf6864be 100644
--- a/cpp/src/qpid/sys/rdma/rdma_wrap.h
+++ b/cpp/src/qpid/sys/rdma/rdma_wrap.h
@@ -58,7 +58,7 @@ namespace Rdma {
 
     private:
         Buffer(uint32_t lkey, char* bytes, const int32_t byteCount);
-        const int32_t bufferSize;
+        int32_t bufferSize;
         ::ibv_sge sge;
     };
 
@@ -124,9 +124,10 @@ namespace Rdma {
         boost::shared_ptr< ::ibv_qp > qp;
         int outstandingSendEvents;
         int outstandingRecvEvents;
-        boost::ptr_deque<Buffer> buffers;
-        qpid::sys::Mutex bufferQueueLock;
-        std::vector<Buffer*> bufferQueue;
+        std::vector<Buffer> sendBuffers;
+        std::vector<Buffer> recvBuffers;
+        qpid::sys::Mutex bufferLock;
+        std::vector<int> freeBuffers;
 
         QueuePair(boost::shared_ptr< ::rdma_cm_id > id);
         ~QueuePair();