PollableQueue: fix unsafe use of deque

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

3 files changed, 13 insertions, 12 deletions

diff --git a/cpp/src/qpid/cluster/Cpg.cpp b/cpp/src/qpid/cluster/Cpg.cpp
index 6a9c97139a..0d7b15cfa8 100644
--- a/cpp/src/qpid/cluster/Cpg.cpp
+++ b/cpp/src/qpid/cluster/Cpg.cpp
@@ -103,11 +103,6 @@ bool Cpg::isFlowControlEnabled() {
 }
 
 bool Cpg::mcast(const iovec* iov, int iovLen) {
-    // Thread-safety note : the cpg_ calls are thread safe, but there
-    // is a race below between calling cpg_flow_control_state_get()
-    // and calling mcast_joined() where N threads could see the state
-    // as disabled and call mcast, but only M < N messages can be sent
-    // without exceeding flow control limits.
     if (isFlowControlEnabled()) {
         QPID_LOG(warning, "CPG flow control enabled")
         return false;
@@ -135,13 +130,13 @@ void Cpg::dispatch(cpg_dispatch_t type) {
 string Cpg::errorStr(cpg_error_t err, const std::string& msg) {
     switch (err) {
       case CPG_OK: return msg+": ok";
-      case CPG_ERR_LIBRARY: return msg+": library";
+      case CPG_ERR_LIBRARY: return msg+": library error";
       case CPG_ERR_TIMEOUT: return msg+": timeout";
       case CPG_ERR_TRY_AGAIN: return msg+": timeout. The aisexec daemon may not be running";
       case CPG_ERR_INVALID_PARAM: return msg+": invalid param";
       case CPG_ERR_NO_MEMORY: return msg+": no memory";
       case CPG_ERR_BAD_HANDLE: return msg+": bad handle";
-      case CPG_ERR_ACCESS: return msg+": access denied. You may need to set your group ID to 'ais'";
+      case CPG_ERR_ACCESS: return msg+": access denied.";
       case CPG_ERR_NOT_EXIST: return msg+": not exist";
       case CPG_ERR_EXIST: return msg+": exist";
       case CPG_ERR_NOT_SUPPORTED: return msg+": not supported";
diff --git a/cpp/src/qpid/cluster/Event.cpp b/cpp/src/qpid/cluster/Event.cpp
index 87cc7e7bd3..b787ef0f67 100644
--- a/cpp/src/qpid/cluster/Event.cpp
+++ b/cpp/src/qpid/cluster/Event.cpp
@@ -39,7 +39,8 @@ Event::Event(EventType t, const ConnectionId& c,  size_t s, uint32_t i)
 
 Event Event::delivered(const MemberId& m, void* d, size_t s) {
     Buffer buf(static_cast<char*>(d), s);
-    EventType type((EventType)buf.getOctet()); 
+    EventType type((EventType)buf.getOctet());
+    assert(type == DATA || type == CONTROL);
     ConnectionId connection(m, reinterpret_cast<Connection*>(buf.getLongLong()));
     uint32_t id = buf.getLong();
     assert(buf.getPosition() == OVERHEAD);
@@ -62,6 +63,7 @@ bool Event::mcast (Cpg& cpg) const {
     b.putOctet(type);
     b.putLongLong(reinterpret_cast<uint64_t>(connectionId.getPointer()));
     b.putLong(id);
+    assert(buf.getPosition() == OVERHEAD);
     iovec iov[] = { { header, OVERHEAD }, { const_cast<char*>(getData()), getSize() } };
     return cpg.mcast(iov, sizeof(iov)/sizeof(*iov));
 }
diff --git a/cpp/src/qpid/sys/PollableQueue.h b/cpp/src/qpid/sys/PollableQueue.h
index 2ee29db022..953d198fb0 100644
--- a/cpp/src/qpid/sys/PollableQueue.h
+++ b/cpp/src/qpid/sys/PollableQueue.h
@@ -121,14 +121,18 @@ template <class T> void PollableQueue<T>::dispatch(sys::DispatchHandle& h) {
     assert(dispatcher.id() == 0 || dispatcher.id() == Thread::current().id());
     dispatcher = Thread::current();
     while (!stopped && !queue.empty()) {
+        T value = queue.front();
+        queue.pop_front();
         bool ok = false;
         {   // unlock to allow concurrent push or call to stop() in callback.
             ScopedUnlock u(lock);
-            // FIXME aconway 2008-12-02: exception-safe if callback throws.
-            ok = callback(queue.front());
+            // FIXME aconway 2008-12-02: not exception safe if callback throws.
+            ok = callback(value);
+        }
+        if (!ok) { // callback cannot process value,  put it back.
+            queue.push_front(value);
+            stopped=true;
         }
-        if (ok) queue.pop_front();
-        else stopped=true;
     }
     dispatcher = Thread();
     if (queue.empty()) condition.clear();