svn commit: r591026 - in /incubator/qpid/trunk/qpid/cpp/src: qpid/framing/SessionState.cpp qpid/framing/SessionState.h tests/SessionState.cpp

[gsim]

Author: gsim
Date: Thu Nov  1 07:44:30 2007
New Revision: 591026

URL: http://svn.apache.org/viewvc?rev=591026&view=rev
Log:
locking around access to the unacked out buffer


Modified:
    incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.cpp
    incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.h
    incubator/qpid/trunk/qpid/cpp/src/tests/SessionState.cpp

Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.cpp
URL: 
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.cpp?rev=591026&r1=591025&r2=591026&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.cpp Thu Nov  1 
07:44:30 2007
@@ -80,8 +80,10 @@
 bool SessionState::sent(const AMQFrame& f) {
     if (isSessionCommand(f))
         return false;
-    if (resumable)
+    if (resumable) {
+        sys::Mutex::ScopedLock l(unackedLock);
         unackedOut.push_back(f);
+    }
     ++lastSent;
     QPID_LOG(trace, "Sent # "<< lastSent << " " << id);
     return ackInterval &&
@@ -91,6 +93,7 @@
 }
 
 SessionState::Replay SessionState::replay() {
+    sys::Mutex::ScopedLock l(unackedLock);
     Replay r(unackedOut.size());
     std::copy(unackedOut.begin(), unackedOut.end(), r.begin());
     return r;
@@ -102,8 +105,10 @@
      if (lastSent < acked)
         throw InvalidArgumentException("Invalid sequence number in ack");
     size_t keep = lastSent - acked;
-    if (keep < unackedOut.size()) 
+    if (keep < unackedOut.size()) {
+        sys::Mutex::ScopedLock l(unackedLock);
         unackedOut.erase(unackedOut.begin(), unackedOut.end()-keep);
+    }
     solicitAckAt = std::max(solicitAckAt, SequenceNumber(acked+ackInterval));
 }
 

Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.h
URL: 
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.h?rev=591026&r1=591025&r2=591026&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.h Thu Nov  1 
07:44:30 2007
@@ -25,6 +25,7 @@
 #include "qpid/framing/SequenceNumber.h"
 #include "qpid/framing/Uuid.h"
 #include "qpid/framing/AMQFrame.h"
+#include "qpid/sys/Mutex.h"
 
 #include <boost/optional.hpp>
 
@@ -127,6 +128,7 @@
     bool ackSolicited;
     bool suspending;
     bool resumable;
+    sys::Mutex unackedLock;
 };
 
 

Modified: incubator/qpid/trunk/qpid/cpp/src/tests/SessionState.cpp
URL: 
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/SessionState.cpp?rev=591026&r1=591025&r2=591026&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/SessionState.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/SessionState.cpp Thu Nov  1 
07:44:30 2007
@@ -49,7 +49,7 @@
 // Received chars as frames
 void received(SessionState& session, const std::string& frames) {
     for_each(frames.begin(), frames.end(),
-             bind(&SessionState::received, session, bind(frame, _1)));
+             bind(&SessionState::received, ref(session), bind(frame, _1)));
 }
 
 // Make a string from a ReplayRange.