QPID-3227: account for extra header bytes when computing data size for send buffers.

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

3 files changed, 14 insertions, 11 deletions

diff --git a/qpid/cpp/src/qpid/sys/rdma/RdmaIO.cpp b/qpid/cpp/src/qpid/sys/rdma/RdmaIO.cpp
index c80c94cba6..af58b7deb0 100644
--- a/qpid/cpp/src/qpid/sys/rdma/RdmaIO.cpp
+++ b/qpid/cpp/src/qpid/sys/rdma/RdmaIO.cpp
@@ -141,7 +141,7 @@ namespace Rdma {
         qp->allocateRecvBuffers(recvBufferCount, bufferSize+FrameHeaderSize);
 
         // Create xmit buffers
-        qp->createSendBuffers(xmitBufferCount, bufferSize+FrameHeaderSize);
+        qp->createSendBuffers(xmitBufferCount, bufferSize, FrameHeaderSize);
     }
 
     AsynchIO::~AsynchIO() {
diff --git a/qpid/cpp/src/qpid/sys/rdma/rdma_wrap.cpp b/qpid/cpp/src/qpid/sys/rdma/rdma_wrap.cpp
index 6d38c42502..efe454c5be 100644
--- a/qpid/cpp/src/qpid/sys/rdma/rdma_wrap.cpp
+++ b/qpid/cpp/src/qpid/sys/rdma/rdma_wrap.cpp
@@ -50,8 +50,9 @@ namespace Rdma {
         return count;
     }
 
-    Buffer::Buffer(uint32_t lkey, char* bytes, const int32_t byteCount) :
-        bufferSize(byteCount)
+    Buffer::Buffer(uint32_t lkey, char* bytes, const int32_t byteCount,
+                   const int32_t reserve) :
+        bufferSize(byteCount + reserve), reserved(reserve)
     {
         sge.addr = (uintptr_t) bytes;
         sge.length = 0;
@@ -163,21 +164,21 @@ namespace Rdma {
     }
 
     // Create buffers to use for writing
-    void QueuePair::createSendBuffers(int sendBufferCount, int bufferSize)
+    void QueuePair::createSendBuffers(int sendBufferCount, int bufferSize, int reserved)
     {
         assert(!smr);
 
         // Round up buffersize to cacheline (64 bytes)
-        bufferSize = (bufferSize+63) & (~63);
+        int dataLength = (bufferSize+reserved+63) & (~63);
 
         // Allocate memory block for all receive buffers
-        char* mem = new char [sendBufferCount * bufferSize];
-        smr = regMr(pd.get(), mem, sendBufferCount * bufferSize, ::IBV_ACCESS_LOCAL_WRITE);
+        char* mem = new char [sendBufferCount * dataLength];
+        smr = regMr(pd.get(), mem, sendBufferCount * dataLength, ::IBV_ACCESS_LOCAL_WRITE);
         sendBuffers.reserve(sendBufferCount);
         freeBuffers.reserve(sendBufferCount);
         for (int i = 0; i<sendBufferCount; ++i) {
             // Allocate xmit buffer
-            sendBuffers.push_back(Buffer(smr->lkey, &mem[i*bufferSize], bufferSize));
+            sendBuffers.push_back(Buffer(smr->lkey, &mem[i*dataLength], bufferSize, reserved));
             freeBuffers.push_back(i);
         }
     }
diff --git a/qpid/cpp/src/qpid/sys/rdma/rdma_wrap.h b/qpid/cpp/src/qpid/sys/rdma/rdma_wrap.h
index 28bddd2165..3737664bde 100644
--- a/qpid/cpp/src/qpid/sys/rdma/rdma_wrap.h
+++ b/qpid/cpp/src/qpid/sys/rdma/rdma_wrap.h
@@ -57,8 +57,9 @@ namespace Rdma {
         void dataCount(int32_t);
 
     private:
-        Buffer(uint32_t lkey, char* bytes, const int32_t byteCount);
+        Buffer(uint32_t lkey, char* bytes, const int32_t byteCount, const int32_t reserve=0);
         int32_t bufferSize;
+        int32_t reserved;   // for framing header
         ::ibv_sge sge;
     };
 
@@ -66,8 +67,9 @@ namespace Rdma {
       return (char*) sge.addr;
     }
 
+    /** return the number of bytes available for application data */
     inline int32_t Buffer::byteCount() const {
-        return bufferSize;
+        return bufferSize - reserved;
     }
 
     inline int32_t Buffer::dataCount() const {
@@ -136,7 +138,7 @@ namespace Rdma {
         typedef boost::intrusive_ptr<QueuePair> intrusive_ptr;
 
         // Create a buffers to use for writing
-        void createSendBuffers(int sendBufferCount, int bufferSize);
+        void createSendBuffers(int sendBufferCount, int dataSize, int headerSize);
 
         // Get a send buffer
         Buffer* getSendBuffer();