qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From acon...@apache.org
Subject svn commit: r743416 - in /qpid/trunk/qpid/cpp/src/qpid: cluster/Connection.cpp cluster/ConnectionMap.cpp sys/AsynchIO.h sys/AsynchIOHandler.cpp sys/posix/AsynchIO.cpp sys/windows/AsynchIO.cpp
Date Wed, 11 Feb 2009 17:29:42 GMT
Author: aconway
Date: Wed Feb 11 17:29:42 2009
New Revision: 743416

URL: http://svn.apache.org/viewvc?rev=743416&view=rev
Log:
Fix race condition with read-credit.

Modified:
    qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp
    qpid/trunk/qpid/cpp/src/qpid/cluster/ConnectionMap.cpp
    qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIO.h
    qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp
    qpid/trunk/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp
    qpid/trunk/qpid/cpp/src/qpid/sys/windows/AsynchIO.cpp

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp?rev=743416&r1=743415&r2=743416&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp Wed Feb 11 17:29:42 2009
@@ -210,6 +210,7 @@
                 remainingSize = size - pi.encodedSize();
             } else {
                 QPID_LOG(debug, "Not enough data for protocol header on outgoing clustered
link");
+                giveReadCredit(1); // We're not going to mcast so give read credit now.
                 return 0;
             }
         }

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/ConnectionMap.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/ConnectionMap.cpp?rev=743416&r1=743415&r2=743416&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/ConnectionMap.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/ConnectionMap.cpp Wed Feb 11 17:29:42 2009
@@ -66,6 +66,7 @@
 ConnectionMap::ConnectionPtr ConnectionMap::getLocal(const ConnectionId& id) {
     if (id.getMember() != cluster.getId()) return 0;
     Map::const_iterator i = map.find(id);
+    assert(i != map.end());     // FIXME aconway 2009-02-11: remove or exception.
     return i == map.end() ? 0 : i->second;
 }
 

Modified: qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIO.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIO.h?rev=743416&r1=743415&r2=743416&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIO.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIO.h Wed Feb 11 17:29:42 2009
@@ -139,6 +139,7 @@
     virtual void queueWriteClose() = 0;
     virtual bool writeQueueEmpty() = 0;
     virtual void startReading() = 0;
+    virtual void stopReading() = 0;
     virtual void requestCallback(RequestCallback) = 0;
     virtual BufferBase* getQueuedBuffer() = 0;
 

Modified: qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp?rev=743416&r1=743415&r2=743416&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp Wed Feb 11 17:29:42 2009
@@ -145,8 +145,10 @@
             // Lock and retest credit to make sure we don't race with increasing credit
             ScopedLock<Mutex> l(creditLock);
             assert(readCredit.get() >= 0);
-            if (readCredit.get() == 0)
+            if (readCredit.get() == 0) {
+                aio->stopReading();
                 return false;
+            }
         }
     }
     return true;

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=743416&r1=743415&r2=743416&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp Wed Feb 11 17:29:42 2009
@@ -267,6 +267,7 @@
     virtual void queueWriteClose();
     virtual bool writeQueueEmpty();
     virtual void startReading();
+    virtual void stopReading();
     virtual void requestCallback(RequestCallback);
     virtual BufferBase* getQueuedBuffer();
 
@@ -389,6 +390,10 @@
     DispatchHandle::rewatchRead();
 }
 
+void AsynchIO::stopReading() {
+    DispatchHandle::unwatchRead();
+}
+
 void AsynchIO::requestCallback(RequestCallback callback) {
     // TODO creating a function object every time isn't all that
     // efficient - if this becomes heavily used do something better (what?)
@@ -439,8 +444,7 @@
                 readTotal += rc;
 
                 if (!readCallback(*this, buff)) {
-                    // We were told to flow control reading at this point
-                    h.unwatchRead();
+                    // We have been flow controlled.
                     break;
                 }
 

Modified: qpid/trunk/qpid/cpp/src/qpid/sys/windows/AsynchIO.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/windows/AsynchIO.cpp?rev=743416&r1=743415&r2=743416&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/windows/AsynchIO.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/windows/AsynchIO.cpp Wed Feb 11 17:29:42 2009
@@ -286,6 +286,7 @@
     virtual void queueWriteClose();
     virtual bool writeQueueEmpty();
     virtual void startReading();
+    virtual void stopReading();
     virtual void requestCallback(RequestCallback);
 
     /**
@@ -534,6 +535,15 @@
     return;
 }
 
+// stopReading was added to prevent a race condition with read-credit on Linux.
+// It may or may not be required on windows.
+// 
+// AsynchIOHandler::readbuff() calls stopReading() inside the same
+// critical section that protects startReading() in
+// AsynchIOHandler::giveReadCredit().
+// 
+void AsynchIO::stopReading() {}
+
 // Queue the specified callback for invocation from an I/O thread.
 void AsynchIO::requestCallback(RequestCallback callback) {
     // This method is generally called from a processing thread; transfer



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


Mime
View raw message