qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kgiu...@apache.org
Subject svn commit: r1303960 - in /qpid/branches/0.16/qpid: cpp/src/qpid/broker/ cpp/src/qpid/broker/MessageGroupManager.cpp cpp/src/qpid/broker/MessageGroupManager.h tests/src/py/qpid_tests/broker_0_10/msg_groups.py
Date Thu, 22 Mar 2012 18:28:55 GMT
Author: kgiusti
Date: Thu Mar 22 18:28:55 2012
New Revision: 1303960

URL: http://svn.apache.org/viewvc?rev=1303960&view=rev
Log:
QPID-3899: merge fix to 0.16

Modified:
    qpid/branches/0.16/qpid/cpp/src/qpid/broker/   (props changed)
    qpid/branches/0.16/qpid/cpp/src/qpid/broker/MessageGroupManager.cpp
    qpid/branches/0.16/qpid/cpp/src/qpid/broker/MessageGroupManager.h
    qpid/branches/0.16/qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py

Propchange: qpid/branches/0.16/qpid/cpp/src/qpid/broker/
------------------------------------------------------------------------------
  Merged /qpid/trunk/qpid/cpp/src/qpid/broker:r1303143-1303144,1303381,1303411

Modified: qpid/branches/0.16/qpid/cpp/src/qpid/broker/MessageGroupManager.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/0.16/qpid/cpp/src/qpid/broker/MessageGroupManager.cpp?rev=1303960&r1=1303959&r2=1303960&view=diff
==============================================================================
--- qpid/branches/0.16/qpid/cpp/src/qpid/broker/MessageGroupManager.cpp (original)
+++ qpid/branches/0.16/qpid/cpp/src/qpid/broker/MessageGroupManager.cpp Thu Mar 22 18:28:55
2012
@@ -43,9 +43,18 @@ const std::string MessageGroupManager::q
 const std::string MessageGroupManager::qpidMessageGroupTimestamp("qpid.group_timestamp");
 
 
+/** return an iterator to the message at position, or members.end() if not found */
+MessageGroupManager::GroupState::MessageFifo::iterator
+MessageGroupManager::GroupState::findMsg(const qpid::framing::SequenceNumber &position)
+{
+    MessageState mState(position);
+    MessageFifo::iterator found = std::lower_bound(members.begin(), members.end(), mState);
+    return (found->position == position) ? found : members.end();
+}
+
 void MessageGroupManager::unFree( const GroupState& state )
 {
-    GroupFifo::iterator pos = freeGroups.find( state.members.front() );
+    GroupFifo::iterator pos = freeGroups.find( state.members.front().position );
     assert( pos != freeGroups.end() && pos->second == &state );
     freeGroups.erase( pos );
 }
@@ -60,8 +69,8 @@ void MessageGroupManager::disown( GroupS
 {
     state.owner.clear();
     assert(state.members.size());
-    assert(freeGroups.find(state.members.front()) == freeGroups.end());
-    freeGroups[state.members.front()] = &state;
+    assert(freeGroups.find(state.members.front().position) == freeGroups.end());
+    freeGroups[state.members.front().position] = &state;
 }
 
 MessageGroupManager::GroupState& MessageGroupManager::findGroup( const QueuedMessage&
qm )
@@ -106,7 +115,8 @@ void MessageGroupManager::enqueued( cons
     // @todo KAG optimization - store reference to group state in QueuedMessage
     // issue: const-ness??
     GroupState& state = findGroup(qm);
-    state.members.push_back(qm.position);
+    GroupState::MessageState mState(qm.position);
+    state.members.push_back(mState);
     uint32_t total = state.members.size();
     QPID_LOG( trace, "group queue " << qName <<
               ": added message to group id=" << state.group << " total=" <<
total );
@@ -123,7 +133,9 @@ void MessageGroupManager::acquired( cons
     // @todo KAG  avoid lookup: retrieve direct reference to group state from QueuedMessage
     // issue: const-ness??
     GroupState& state = findGroup(qm);
-    assert(state.members.size());   // there are msgs present
+    GroupState::MessageFifo::iterator m = state.findMsg(qm.position);
+    assert(m != state.members.end());
+    m->acquired = true;
     state.acquired += 1;
     QPID_LOG( trace, "group queue " << qName <<
               ": acquired message in group id=" << state.group << " acquired="
<< state.acquired );
@@ -137,6 +149,9 @@ void MessageGroupManager::requeued( cons
     GroupState& state = findGroup(qm);
     assert( state.acquired != 0 );
     state.acquired -= 1;
+    GroupState::MessageFifo::iterator m = state.findMsg(qm.position);
+    assert(m != state.members.end());
+    m->acquired = false;
     if (state.acquired == 0 && state.owned()) {
         QPID_LOG( trace, "group queue " << qName <<
                   ": consumer name=" << state.owner << " released group id="
<< state.group);
@@ -152,13 +167,17 @@ void MessageGroupManager::dequeued( cons
     // @todo KAG  avoid lookup: retrieve direct reference to group state from QueuedMessage
     // issue: const-ness??
     GroupState& state = findGroup(qm);
-    assert( state.members.size() != 0 );
-    assert( state.acquired != 0 );
-    state.acquired -= 1;
+    GroupState::MessageFifo::iterator m = state.findMsg(qm.position);
+    assert(m != state.members.end());
+    if (m->acquired) {
+        assert( state.acquired != 0 );
+        state.acquired -= 1;
+    }
 
-    // likely to be at or near begin() if dequeued in order
+    // special case if qm is first (oldest) message in the group:
+    // may need to re-insert it back on the freeGroups list, as the index will change
     bool reFreeNeeded = false;
-    if (state.members.front() == qm.position) {
+    if (m == state.members.begin()) {
         if (!state.owned()) {
             // will be on the freeGroups list if mgmt is dequeueing rather than a consumer!
             // if on freelist, it is indexed by first member, which is about to be removed!
@@ -167,15 +186,7 @@ void MessageGroupManager::dequeued( cons
         }
         state.members.pop_front();
     } else {
-        GroupState::PositionFifo::iterator pos = state.members.begin() + 1;
-        GroupState::PositionFifo::iterator end = state.members.end();
-        while (pos != end) {
-            if (*pos == qm.position) {
-                state.members.erase(pos);
-                break;
-            }
-            ++pos;
-        }
+        state.members.erase(m);
     }
 
     uint32_t total = state.members.size();
@@ -220,11 +231,11 @@ bool MessageGroupManager::nextConsumable
         GroupState& group = findGroup(next);
         if (!group.owned()) {
             //TODO: make acquire more efficient when we already have the message in question
-            if (group.members.front() == next.position && messages.acquire(next.position,
next)) {    // only take from head!
+            if (group.members.front().position == next.position && messages.acquire(next.position,
next)) {    // only take from head!
                 return true;
             }
             QPID_LOG(debug, "Skipping " << next.position << " since group " <<
group.group
-                     << "'s head message still pending. pos=" << group.members.front());
+                     << "'s head message still pending. pos=" << group.members.front().position);
         } else if (group.owner == c->getName() && messages.acquire(next.position,
next)) {
             return true;
         }
@@ -284,7 +295,7 @@ void MessageGroupManager::query(qpid::ty
         info[GROUP_TIMESTAMP] = 0;
         if (g->second.members.size() != 0) {
             QueuedMessage qm;
-            if (messages.find(g->second.members.front(), qm) &&
+            if (messages.find(g->second.members.front().position, qm) &&
                 qm.payload &&
                 qm.payload->hasProperties<framing::DeliveryProperties>()) {
                 info[GROUP_TIMESTAMP] = qm.payload->getProperties<framing::DeliveryProperties>()->getTimestamp();
@@ -353,6 +364,7 @@ namespace {
     const std::string GROUP_OWNER("owner");
     const std::string GROUP_ACQUIRED_CT("acquired-ct");
     const std::string GROUP_POSITIONS("positions");
+    const std::string GROUP_ACQUIRED_MSGS("acquired-msgs");
     const std::string GROUP_STATE("group-state");
 }
 
@@ -371,10 +383,14 @@ void MessageGroupManager::getState(qpid:
         group.setString(GROUP_OWNER, g->second.owner);
         group.setInt(GROUP_ACQUIRED_CT, g->second.acquired);
         framing::Array positions(TYPE_CODE_UINT32);
-        for (GroupState::PositionFifo::const_iterator p = g->second.members.begin();
-             p != g->second.members.end(); ++p)
-            positions.push_back(framing::Array::ValuePtr(new IntegerValue( *p )));
+        framing::Array acquiredMsgs(TYPE_CODE_BOOLEAN);
+        for (GroupState::MessageFifo::const_iterator p = g->second.members.begin();
+             p != g->second.members.end(); ++p) {
+            positions.push_back(framing::Array::ValuePtr(new IntegerValue( p->position
)));
+            acquiredMsgs.push_back(framing::Array::ValuePtr(new BoolValue( p->acquired
)));
+        }
         group.setArray(GROUP_POSITIONS, positions);
+        group.setArray(GROUP_ACQUIRED_MSGS, acquiredMsgs);
         groupState.push_back(framing::Array::ValuePtr(new FieldTableValue(group)));
     }
     state.setArray(GROUP_STATE, groupState);
@@ -425,13 +441,25 @@ void MessageGroupManager::setState(const
                      qName << "\": position encoding error!");
             return;
         }
+        framing::Array acquiredMsgs(TYPE_CODE_BOOLEAN);
+        ok = group.getArray(GROUP_ACQUIRED_MSGS, acquiredMsgs);
+        if (!ok || positions.count() != acquiredMsgs.count()) {
+            QPID_LOG(error, "Invalid message group state information for queue \"" <<
+                     qName << "\": acquired flag encoding error!");
+            return;
+        }
+
+        Array::const_iterator a = acquiredMsgs.begin();
+        for (Array::const_iterator p = positions.begin(); p != positions.end(); ++p) {
+            GroupState::MessageState mState((*p)->getIntegerValue<uint32_t, 4>());
+            mState.acquired = (*a++)->getIntegerValue<bool>();
+            state.members.push_back(mState);
+        }
 
-        for (Array::const_iterator p = positions.begin(); p != positions.end(); ++p)
-            state.members.push_back((*p)->getIntegerValue<uint32_t, 4>());
         messageGroups[state.group] = state;
         if (!state.owned()) {
             assert(state.members.size());
-            freeGroups[state.members.front()] = &messageGroups[state.group];
+            freeGroups[state.members.front().position] = &messageGroups[state.group];
         }
     }
 

Modified: qpid/branches/0.16/qpid/cpp/src/qpid/broker/MessageGroupManager.h
URL: http://svn.apache.org/viewvc/qpid/branches/0.16/qpid/cpp/src/qpid/broker/MessageGroupManager.h?rev=1303960&r1=1303959&r2=1303960&view=diff
==============================================================================
--- qpid/branches/0.16/qpid/cpp/src/qpid/broker/MessageGroupManager.h (original)
+++ qpid/branches/0.16/qpid/cpp/src/qpid/broker/MessageGroupManager.h Thu Mar 22 18:28:55
2012
@@ -45,19 +45,29 @@ class MessageGroupManager : public State
 
     struct GroupState {
         // note: update getState()/setState() when changing this object's state implementation
-        typedef std::deque<framing::SequenceNumber> PositionFifo;
+
+        // track which messages are in this group, and if they have been acquired
+        struct MessageState {
+            qpid::framing::SequenceNumber position;
+            bool                    acquired;
+            MessageState() : acquired(false) {}
+            MessageState(const qpid::framing::SequenceNumber& p) : position(p), acquired(false)
{}
+            bool operator<(const MessageState& b) const { return position < b.position;
}
+        };
+        typedef std::deque<MessageState> MessageFifo;
 
         std::string group;  // group identifier
         std::string owner;  // consumer with outstanding acquired messages
         uint32_t acquired;  // count of outstanding acquired messages
-        PositionFifo members;   // msgs belonging to this group
+        MessageFifo members;   // msgs belonging to this group, in enqueue order
 
         GroupState() : acquired(0) {}
         bool owned() const {return !owner.empty();}
+        MessageFifo::iterator findMsg(const qpid::framing::SequenceNumber &);
     };
 
     typedef sys::unordered_map<std::string, struct GroupState> GroupMap;
-    typedef std::map<framing::SequenceNumber, struct GroupState *> GroupFifo;
+    typedef std::map<qpid::framing::SequenceNumber, struct GroupState *> GroupFifo;
 
     GroupMap messageGroups; // index: group name
     GroupFifo freeGroups;   // ordered by oldest free msg

Modified: qpid/branches/0.16/qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py
URL: http://svn.apache.org/viewvc/qpid/branches/0.16/qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py?rev=1303960&r1=1303959&r2=1303960&view=diff
==============================================================================
--- qpid/branches/0.16/qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py (original)
+++ qpid/branches/0.16/qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py Thu Mar 22 18:28:55
2012
@@ -1122,6 +1122,70 @@ class MultiConsumerMsgGroupTests(Base):
         snd.close()
 
 
+    def test_ttl_expire(self):
+        """ Verify that expired (TTL) group messages are skipped correctly
+        """
+        snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," +
+                              " node: {x-declare: {arguments:" +
+                              " {'qpid.group_header_key':'THE-GROUP'," +
+                              "'qpid.shared_msg_group':1}}}}")
+
+        groups = ["A","B","C","A","B","C"]
+        messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups]
+        index = 0
+        for m in messages:
+            m.content['index'] = index
+            index += 1
+            if m.properties['THE-GROUP'] == 'B':
+                m.ttl = 1;
+            snd.send(m)
+
+        sleep(2)  # let all B's expire
+
+        # create consumers on separate sessions: C1,C2
+        s1 = self.setup_session()
+        c1 = s1.receiver("msg-group-q", options={"capacity":0})
+        s2 = self.setup_session()
+        c2 = s2.receiver("msg-group-q", options={"capacity":0})
+
+        # C1 should acquire A-0, then C2 should acquire C-2, Group B should
+        # expire and never be fetched
+
+        m1 = c1.fetch(0);
+        assert m1.properties['THE-GROUP'] == 'A'
+        assert m1.content['index'] == 0
+
+        m2 = c2.fetch(0);
+        assert m2.properties['THE-GROUP'] == 'C'
+        assert m2.content['index'] == 2
+
+        m1 = c1.fetch(0);
+        assert m1.properties['THE-GROUP'] == 'A'
+        assert m1.content['index'] == 3
+
+        m2 = c2.fetch(0);
+        assert m2.properties['THE-GROUP'] == 'C'
+        assert m2.content['index'] == 5
+
+        # there should be no more left for either consumer
+        try:
+            mx = c1.fetch(0)
+            assert False     # should never get here
+        except Empty:
+            pass
+        try:
+            mx = c2.fetch(0)
+            assert False     # should never get here
+        except Empty:
+            pass
+
+        c1.session.acknowledge()
+        c2.session.acknowledge()
+        c1.close()
+        c2.close()
+        snd.close()
+
+
 class StickyConsumerMsgGroupTests(Base):
     """
     Tests for the behavior of sticky-consumer message groups.  These tests



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org


Mime
View raw message