qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From astitc...@apache.org
Subject svn commit: r1200925 - in /qpid/trunk/qpid/cpp/src/qpid/sys: posix/AsynchIO.cpp ssl/SslIo.cpp
Date Fri, 11 Nov 2011 16:06:08 GMT
Author: astitcher
Date: Fri Nov 11 16:06:08 2011
New Revision: 1200925

URL: http://svn.apache.org/viewvc?rev=1200925&view=rev
Log:
QPID-3608: Improve C++ broker consume performance

Modified:
    qpid/trunk/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp
    qpid/trunk/qpid/cpp/src/qpid/sys/ssl/SslIo.cpp

Modified: qpid/trunk/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp?rev=1200925&r1=1200924&r2=1200925&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp Fri Nov 11 16:06:08 2011
@@ -61,11 +61,10 @@ struct StaticInit {
  * case we could rebalance the info occasionally.  
  */
 __thread int threadReadTotal = 0;
-__thread int threadMaxRead = 0;
 __thread int threadReadCount = 0;
 __thread int threadWriteTotal = 0;
 __thread int threadWriteCount = 0;
-__thread int64_t threadMaxReadTimeNs = 2 * 1000000; // start at 2ms
+__thread int64_t threadMaxIoTimeNs = 2 * 1000000; // start at 2ms
 }
 
 /*
@@ -426,7 +425,6 @@ void AsynchIO::readable(DispatchHandle& 
         // We have been flow controlled.
         return;
     }
-    int readTotal = 0;
     AbsTime readStartTime = AbsTime::now();
     do {
         // (Try to) get a buffer
@@ -441,7 +439,6 @@ void AsynchIO::readable(DispatchHandle& 
             if (rc > 0) {
                 buff->dataCount += rc;
                 threadReadTotal += rc;
-                readTotal += rc;
 
                 readCallback(*this, buff);
                 if (readingStopped) {
@@ -453,17 +450,17 @@ void AsynchIO::readable(DispatchHandle& 
                     // If we didn't fill the read buffer then time to stop reading
                     break;
                 }
-                
+
                 // Stop reading if we've overrun our timeslot
-                if (Duration(readStartTime, AbsTime::now()) > threadMaxReadTimeNs) {
+                if (Duration(readStartTime, AbsTime::now()) > threadMaxIoTimeNs) {
                     break;
                 }
-                
+
             } else {
                 // Put buffer back (at front so it doesn't interfere with unread buffers)
                 bufferQueue.push_front(buff);
                 assert(buff);
-                
+
                 // Eof or other side has gone away
                 if (rc == 0 || errno == ECONNRESET) {
                     eofCallback(*this);
@@ -491,12 +488,11 @@ void AsynchIO::readable(DispatchHandle& 
                 h.unwatchRead();
                 break;
             }
-            
+
         }
     } while (true);
 
     ++threadReadCount;
-    threadMaxRead = std::max(threadMaxRead, readTotal);
     return;
 }
 
@@ -504,7 +500,7 @@ void AsynchIO::readable(DispatchHandle& 
  * We carry on writing whilst we have data to write and we can write
  */
 void AsynchIO::writeable(DispatchHandle& h) {
-    int writeTotal = 0;
+    AbsTime writeStartTime = AbsTime::now();
     do {
         // See if we've got something to write
         if (!writeQueue.empty()) {
@@ -516,7 +512,6 @@ void AsynchIO::writeable(DispatchHandle&
             int rc = socket.write(buff->bytes+buff->dataStart, buff->dataCount);
             if (rc >= 0) {
                 threadWriteTotal += rc;
-                writeTotal += rc;
 
                 // If we didn't write full buffer put rest back
                 if (rc != buff->dataCount) {
@@ -525,14 +520,14 @@ void AsynchIO::writeable(DispatchHandle&
                     writeQueue.push_back(buff);
                     break;
                 }
-                
+
                 // Recycle the buffer
                 queueReadBuffer(buff);
-                
-                // If we've already written more than the max for reading then stop
-                // (this is to stop writes dominating reads) 
-                if (writeTotal > threadMaxRead)
+
+                // Stop writing if we've overrun our timeslot
+                if (Duration(writeStartTime, AbsTime::now()) > threadMaxIoTimeNs) {
                     break;
+                }
             } else {
                 // Put buffer back
                 writeQueue.push_back(buff);
@@ -580,7 +575,7 @@ void AsynchIO::writeable(DispatchHandle&
     ++threadWriteCount;
     return;
 }
-        
+
 void AsynchIO::disconnected(DispatchHandle& h) {
     // If we have not already queued close then call disconnected callback before closing
     if (!queuedClose && disCallback) disCallback(*this);

Modified: qpid/trunk/qpid/cpp/src/qpid/sys/ssl/SslIo.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/ssl/SslIo.cpp?rev=1200925&r1=1200924&r2=1200925&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/ssl/SslIo.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/ssl/SslIo.cpp Fri Nov 11 16:06:08 2011
@@ -57,11 +57,10 @@ void ignoreSigpipe() {
  * case we could rebalance the info occasionally.  
  */
 __thread int threadReadTotal = 0;
-__thread int threadMaxRead = 0;
 __thread int threadReadCount = 0;
 __thread int threadWriteTotal = 0;
 __thread int threadWriteCount = 0;
-__thread int64_t threadMaxReadTimeNs = 2 * 1000000; // start at 2ms
+__thread int64_t threadMaxIoTimeNs = 2 * 1000000; // start at 2ms
 }
 
 /*
@@ -277,7 +276,6 @@ SslIO::BufferBase* SslIO::getQueuedBuffe
  * it in
  */
 void SslIO::readable(DispatchHandle& h) {
-    int readTotal = 0;
     AbsTime readStartTime = AbsTime::now();
     do {
         // (Try to) get a buffer
@@ -292,24 +290,23 @@ void SslIO::readable(DispatchHandle& h) 
             if (rc > 0) {
                 buff->dataCount += rc;
                 threadReadTotal += rc;
-                readTotal += rc;
 
                 readCallback(*this, buff);
                 if (rc != readCount) {
                     // If we didn't fill the read buffer then time to stop reading
                     break;
                 }
-                
+
                 // Stop reading if we've overrun our timeslot
-                if (Duration(readStartTime, AbsTime::now()) > threadMaxReadTimeNs) {
+                if (Duration(readStartTime, AbsTime::now()) > threadMaxIoTimeNs) {
                     break;
                 }
-                
+
             } else {
                 // Put buffer back (at front so it doesn't interfere with unread buffers)
                 bufferQueue.push_front(buff);
                 assert(buff);
-                
+
                 // Eof or other side has gone away
                 if (rc == 0 || errno == ECONNRESET) {
                     eofCallback(*this);
@@ -337,12 +334,11 @@ void SslIO::readable(DispatchHandle& h) 
                 h.unwatchRead();
                 break;
             }
-            
+
         }
     } while (true);
 
     ++threadReadCount;
-    threadMaxRead = std::max(threadMaxRead, readTotal);
     return;
 }
 
@@ -350,7 +346,7 @@ void SslIO::readable(DispatchHandle& h) 
  * We carry on writing whilst we have data to write and we can write
  */
 void SslIO::writeable(DispatchHandle& h) {
-    int writeTotal = 0;
+    AbsTime writeStartTime = AbsTime::now();
     do {
         // See if we've got something to write
         if (!writeQueue.empty()) {
@@ -362,7 +358,6 @@ void SslIO::writeable(DispatchHandle& h)
             int rc = socket.write(buff->bytes+buff->dataStart, buff->dataCount);
             if (rc >= 0) {
                 threadWriteTotal += rc;
-                writeTotal += rc;
 
                 // If we didn't write full buffer put rest back
                 if (rc != buff->dataCount) {
@@ -371,14 +366,14 @@ void SslIO::writeable(DispatchHandle& h)
                     writeQueue.push_back(buff);
                     break;
                 }
-                
+
                 // Recycle the buffer
                 queueReadBuffer(buff);
-                
-                // If we've already written more than the max for reading then stop
-                // (this is to stop writes dominating reads) 
-                if (writeTotal > threadMaxRead)
+
+                // Stop writing if we've overrun our timeslot
+                if (Duration(writeStartTime, AbsTime::now()) > threadMaxIoTimeNs) {
                     break;
+                }
             } else {
                 // Put buffer back
                 writeQueue.push_back(buff);
@@ -425,7 +420,7 @@ void SslIO::writeable(DispatchHandle& h)
     ++threadWriteCount;
     return;
 }
-        
+
 void SslIO::disconnected(DispatchHandle& h) {
     // If we've already queued close do it instead of disconnected callback
     if (queuedClose) {



---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project:      http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org


Mime
View raw message