You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2013/08/05 21:33:36 UTC
svn commit: r1510678 - in /qpid/trunk/qpid/cpp/src: qpid/ha/ tests/
Author: aconway
Date: Mon Aug 5 19:33:35 2013
New Revision: 1510678
URL: http://svn.apache.org/r1510678
Log:
QPID-4327: HA Handle brokers joining and leaving during a transaction.
During a transaction:
- A broker leaving aborts the transaction.
- A broker joining does not participate in the transaction
- but does receive the results of the TX via normal replication.
Clean up tx-queues when the transaction completes.
Modified:
qpid/trunk/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp
qpid/trunk/qpid/cpp/src/qpid/ha/Event.cpp
qpid/trunk/qpid/cpp/src/qpid/ha/Event.h
qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp
qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.h
qpid/trunk/qpid/cpp/src/qpid/ha/Membership.cpp
qpid/trunk/qpid/cpp/src/qpid/ha/Membership.h
qpid/trunk/qpid/cpp/src/qpid/ha/Primary.cpp
qpid/trunk/qpid/cpp/src/qpid/ha/PrimaryTxObserver.cpp
qpid/trunk/qpid/cpp/src/qpid/ha/PrimaryTxObserver.h
qpid/trunk/qpid/cpp/src/qpid/ha/QueueReplicator.cpp
qpid/trunk/qpid/cpp/src/qpid/ha/ReplicatingSubscription.h
qpid/trunk/qpid/cpp/src/qpid/ha/TxReplicator.cpp
qpid/trunk/qpid/cpp/src/qpid/ha/TxReplicator.h
qpid/trunk/qpid/cpp/src/qpid/ha/types.cpp
qpid/trunk/qpid/cpp/src/qpid/ha/types.h
qpid/trunk/qpid/cpp/src/tests/ha_test.py
qpid/trunk/qpid/cpp/src/tests/ha_tests.py
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp?rev=1510678&r1=1510677&r2=1510678&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp Mon Aug 5 19:33:35 2013
@@ -793,7 +793,7 @@ void BrokerReplicator::deleteQueue(const
// messages. Any reroutes will be done at the primary and
// replicated as normal.
if (purge) queue->purge(0, boost::shared_ptr<Exchange>());
- broker.deleteQueue(name, userId, remoteHost);
+ haBroker.deleteQueue(name, remoteHost);
QPID_LOG(debug, logPrefix << "Queue deleted: " << name);
}
}
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/Event.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/Event.cpp?rev=1510678&r1=1510677&r2=1510678&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/Event.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/Event.cpp Mon Aug 5 19:33:35 2013
@@ -44,13 +44,14 @@ bool isEventKey(const std::string& key)
const string DequeueEvent::KEY(QPID_HA+"de");
const string IdEvent::KEY(QPID_HA+"id");
-const string TxEnqueueEvent::KEY(QPID_HA+"txen");
-const string TxDequeueEvent::KEY(QPID_HA+"txde");
-const string TxPrepareEvent::KEY(QPID_HA+"txpr");
-const string TxCommitEvent::KEY(QPID_HA+"txcm");
+const string TxEnqueueEvent::KEY(QPID_HA+"txenq");
+const string TxDequeueEvent::KEY(QPID_HA+"txdeq");
+const string TxPrepareEvent::KEY(QPID_HA+"txpre");
+const string TxCommitEvent::KEY(QPID_HA+"txcom");
const string TxRollbackEvent::KEY(QPID_HA+"txrb");
const string TxPrepareOkEvent::KEY(QPID_HA+"txok");
const string TxPrepareFailEvent::KEY(QPID_HA+"txno");
+const string TxMembersEvent::KEY(QPID_HA+"txmem");
broker::Message makeMessage(
const string& data, const string& destination, const string& routingKey)
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/Event.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/Event.h?rev=1510678&r1=1510677&r2=1510678&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/Event.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/Event.h Mon Aug 5 19:33:35 2013
@@ -43,7 +43,8 @@ broker::Message makeMessage(
bool isEventKey(const std::string& key);
/** Base class for encodable events */
-struct Event {
+class Event {
+ public:
virtual ~Event() {}
virtual void encode(framing::Buffer& buffer) const = 0;
virtual void decode(framing::Buffer& buffer) = 0;
@@ -62,7 +63,8 @@ inline std::ostream& operator<<(std::ost
}
/** Event base template */
-template <class Derived> struct EventBase : public Event {
+template <class Derived> class EventBase : public Event {
+ public:
std::string key() const { return Derived::KEY; }
};
@@ -176,6 +178,16 @@ struct TxPrepareFailEvent : public Event
void print(std::ostream& o) const { o << broker; }
};
+struct TxMembersEvent : public EventBase<TxMembersEvent> {
+ static const std::string KEY;
+ UuidSet members;
+ TxMembersEvent(const UuidSet& s=UuidSet()) : members(s) {}
+ void encode(framing::Buffer& b) const { b.put(members); }
+ void decode(framing::Buffer& b) { b.get(members); }
+ size_t encodedSize() const { return members.encodedSize(); }
+ void print(std::ostream& o) const { o << members; }
+};
+
}} // namespace qpid::ha
#endif /*!QPID_HA_EVENT_H*/
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp?rev=1510678&r1=1510677&r2=1510678&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp Mon Aug 5 19:33:35 2013
@@ -221,4 +221,8 @@ boost::shared_ptr<QueueReplicator> HaBro
broker.getExchanges().find(QueueReplicator::replicatorName(queueName)));
}
+void HaBroker::deleteQueue(const string& name, const string& connectionId) {
+ broker.deleteQueue(name, settings.username, connectionId);
+}
+
}} // namespace qpid::ha
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.h?rev=1510678&r1=1510677&r2=1510678&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.h Mon Aug 5 19:33:35 2013
@@ -102,6 +102,10 @@ class HaBroker : public management::Mana
boost::shared_ptr<QueueReplicator> findQueueReplicator(const std::string& queueName);
+ /**@param connectionId optional, only available on backup */
+ void deleteQueue(const std::string& name,
+ const std::string& connectionId=std::string());
+
private:
void setPublicUrl(const Url&);
void setBrokerUrl(const Url&);
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/Membership.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/Membership.cpp?rev=1510678&r1=1510677&r2=1510678&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/Membership.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/Membership.cpp Mon Aug 5 19:33:35 2013
@@ -144,9 +144,19 @@ bool checkTransition(BrokerStatus from,
}
} // namespace
+void Membership::addCallback(UpdateCallback cb) {
+ Mutex::ScopedLock l(lock);
+ callbacks.push_back(cb);
+ cb(brokers); // Give an initial update.
+}
void Membership::update(Mutex::ScopedLock& l) {
QPID_LOG(info, "Membership: " << brokers);
+
+ // Call callbacks
+ for_each(callbacks.begin(), callbacks.end(),
+ boost::bind<void>(&UpdateCallback::operator(), _1, boost::cref(brokers)));
+
// Update managment and send update event.
BrokerStatus newStatus = getStatus(l);
Variant::List brokerList = asList(l);
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/Membership.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/Membership.h?rev=1510678&r1=1510677&r2=1510678&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/Membership.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/Membership.h Mon Aug 5 19:33:35 2013
@@ -61,6 +61,11 @@ class Membership
void setMgmtObject(boost::shared_ptr<qmf::org::apache::qpid::ha::HaBroker>);
+ /** Call callback when membership changes.
+ * NOTE: called with Membership lock held.
+ */
+ typedef boost::function<void(const BrokerInfo::Map&)> UpdateCallback;
+ void addCallback(UpdateCallback);
void clear(); ///< Clear all but self.
void add(const BrokerInfo& b);
void remove(const types::Uuid& id);
@@ -94,6 +99,7 @@ class Membership
const types::Uuid self;
BrokerInfo::Map brokers;
BrokerStatus oldStatus;
+ std::vector<UpdateCallback> callbacks;
};
}} // namespace qpid::ha
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/Primary.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/Primary.cpp?rev=1510678&r1=1510677&r2=1510678&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/Primary.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/Primary.cpp Mon Aug 5 19:33:35 2013
@@ -301,6 +301,7 @@ void Primary::backupDisconnect(shared_pt
backup->cancel();
expectedBackups.erase(backup);
backups.erase(id);
+ membership.remove(id);
}
@@ -387,14 +388,12 @@ void Primary::setCatchupQueues(const Rem
}
void Primary::startTx(const boost::shared_ptr<broker::TxBuffer>& tx) {
- QPID_LOG(debug, logPrefix << "Started TX transaction");
shared_ptr<PrimaryTxObserver> observer(new PrimaryTxObserver(haBroker));
observer->initialize();
tx->setObserver(observer);
}
void Primary::startDtx(const boost::shared_ptr<broker::DtxBuffer>& dtx) {
- QPID_LOG(debug, logPrefix << "Started DTX transaction");
shared_ptr<PrimaryTxObserver> observer(new PrimaryTxObserver(haBroker));
observer->initialize();
dtx->setObserver(observer);
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/PrimaryTxObserver.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/PrimaryTxObserver.cpp?rev=1510678&r1=1510677&r2=1510678&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/PrimaryTxObserver.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/PrimaryTxObserver.cpp Mon Aug 5 19:33:35 2013
@@ -31,6 +31,7 @@
#include "qpid/broker/Broker.h"
#include "qpid/broker/Queue.h"
#include <boost/lexical_cast.hpp>
+#include <algorithm>
namespace qpid {
namespace framing {
@@ -87,9 +88,11 @@ PrimaryTxObserver::PrimaryTxObserver(HaB
// when the tx-queue is deleted.
//
BrokerInfo::Set infoSet(haBroker.getMembership().otherBackups());
- std::transform(infoSet.begin(), infoSet.end(), inserter(backups, backups.begin()),
- boost::bind(&BrokerInfo::getSystemId, _1));
- QPID_LOG(debug, logPrefix << "Started on " << backups);
+ std::transform(infoSet.begin(), infoSet.end(), inserter(members, members.begin()),
+ boost::bind(&BrokerInfo::getSystemId, _1));
+
+ QPID_LOG(debug, logPrefix << "Started TX " << id);
+ QPID_LOG(debug, logPrefix << "Members: " << members);
pair<QueuePtr, bool> result =
broker.getQueues().declare(
@@ -97,8 +100,15 @@ PrimaryTxObserver::PrimaryTxObserver(HaB
QueueSettings(/*durable*/false, /*autodelete*/true));
assert(result.second);
txQueue = result.first;
+ txQueue->deliver(TxMembersEvent(members).message());
+ // Do this last, it will start concurrent callbacks.
+ haBroker.getMembership().addCallback(
+ boost::bind(&PrimaryTxObserver::membership, this, _1));
}
+PrimaryTxObserver::~PrimaryTxObserver() {}
+
+
void PrimaryTxObserver::initialize() {
broker.getExchanges().registerExchange(
boost::shared_ptr<Exchange>(new Exchange(shared_from_this())));
@@ -125,7 +135,7 @@ void PrimaryTxObserver::deduplicate(sys:
boost::shared_ptr<Primary> primary(boost::dynamic_pointer_cast<Primary>(haBroker.getRole()));
assert(primary);
// Tell replicating subscriptions to skip IDs in the transaction.
- for (UuidSet::iterator b = backups.begin(); b != backups.end(); ++b)
+ for (UuidSet::iterator b = members.begin(); b != members.end(); ++b)
for (QueueIdsMap::iterator q = enqueues.begin(); q != enqueues.end(); ++q)
primary->skip(*b, q->first, q->second);
}
@@ -136,7 +146,8 @@ bool PrimaryTxObserver::prepare() {
QPID_LOG(debug, logPrefix << "Prepare");
deduplicate(l);
txQueue->deliver(TxPrepareEvent().message());
- while (!isPrepared(l)) lock.wait();
+ while (prepared != members && !failed)
+ lock.wait();
return !failed;
}
@@ -144,12 +155,20 @@ void PrimaryTxObserver::commit() {
sys::Mutex::ScopedLock l(lock);
QPID_LOG(debug, logPrefix << "Commit");
txQueue->deliver(TxCommitEvent().message());
+ destroy();
}
void PrimaryTxObserver::rollback() {
sys::Mutex::ScopedLock l(lock);
QPID_LOG(debug, logPrefix << "Rollback");
txQueue->deliver(TxRollbackEvent().message());
+ destroy();
+}
+
+void PrimaryTxObserver::destroy() {
+ // Destroying the queue will result in destruction of this when
+ // the queues observer references are cleared.
+ haBroker.deleteQueue(txQueue->getName());
}
void PrimaryTxObserver::txPrepareOkEvent(const string& data) {
@@ -169,8 +188,15 @@ void PrimaryTxObserver::txPrepareFailEve
lock.notify();
}
-bool PrimaryTxObserver::isPrepared(sys::Mutex::ScopedLock&) {
- return (prepared == backups || failed);
+void PrimaryTxObserver::membership(const BrokerInfo::Map& update) {
+ sys::Mutex::ScopedLock l(lock);
+ for (UuidSet::iterator i = members.begin(); i != members.end(); ++i)
+ if (!update.count(*i)) { // A broker is down
+ failed = true;
+ lock.notify();
+ return;
+ }
}
+
}} // namespace qpid::ha
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/PrimaryTxObserver.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/PrimaryTxObserver.h?rev=1510678&r1=1510677&r2=1510678&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/PrimaryTxObserver.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/PrimaryTxObserver.h Mon Aug 5 19:33:35 2013
@@ -63,6 +63,7 @@ class PrimaryTxObserver : public broker:
{
public:
PrimaryTxObserver(HaBroker&);
+ ~PrimaryTxObserver();
/** Call immediately after constructor, uses shared_from_this. */
void initialize();
@@ -80,11 +81,13 @@ class PrimaryTxObserver : public broker:
typedef qpid::sys::unordered_map<
QueuePtr, ReplicationIdSet, boost::hash<QueuePtr> > QueueIdsMap;
+ void membership(const BrokerInfo::Map&);
void deduplicate(sys::Mutex::ScopedLock&);
void txPrepareOkEvent(const std::string& data);
void txPrepareFailEvent(const std::string& data);
void consumerRemoved(const broker::Consumer&);
bool isPrepared(sys::Mutex::ScopedLock&);
+ void destroy();
sys::Monitor lock;
std::string logPrefix;
@@ -94,7 +97,7 @@ class PrimaryTxObserver : public broker:
QueuePtr txQueue;
QueueIdsMap enqueues;
bool failed;
- UuidSet backups;
+ UuidSet members;
UuidSet prepared;
};
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/QueueReplicator.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/QueueReplicator.cpp?rev=1510678&r1=1510677&r2=1510678&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/QueueReplicator.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/QueueReplicator.cpp Mon Aug 5 19:33:35 2013
@@ -182,7 +182,6 @@ void QueueReplicator::destroy() {
bridge2 = bridge; // call close outside the lock.
// Need to drop shared pointers to avoid pointer cycles keeping this in memory.
queue.reset();
- link.reset();
bridge.reset();
getBroker()->getExchanges().destroy(getName());
}
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/ReplicatingSubscription.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/ReplicatingSubscription.h?rev=1510678&r1=1510677&r2=1510678&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/ReplicatingSubscription.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/ReplicatingSubscription.h Mon Aug 5 19:33:35 2013
@@ -43,7 +43,7 @@ class Buffer;
namespace ha {
class QueueGuard;
class HaBroker;
-struct Event;
+class Event;
class Primary;
/**
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/TxReplicator.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/TxReplicator.cpp?rev=1510678&r1=1510677&r2=1510678&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/TxReplicator.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/TxReplicator.cpp Mon Aug 5 19:33:35 2013
@@ -81,13 +81,11 @@ TxReplicator::TxReplicator(
channel(link->nextChannel()),
dequeueState(hb.getBroker().getQueues())
{
- string shortId = getTxId(txQueue->getName()).substr(0, 8);
+ string id(getTxId(txQueue->getName()));
+ string shortId = id.substr(0, 8);
logPrefix = "Backup of transaction "+shortId+": ";
-
+ QPID_LOG(debug, logPrefix << "Started TX " << id);
if (!store) throw Exception(QPID_MSG(logPrefix << "No message store loaded."));
- boost::shared_ptr<Backup> backup = boost::dynamic_pointer_cast<Backup>(hb.getRole());
- if (!backup) throw Exception(QPID_MSG(logPrefix << "Broker is not in backup mode."));
- brokerReplicator = backup->getBrokerReplicator();
// Dispatch transaction events.
dispatch[TxEnqueueEvent::KEY] =
@@ -100,6 +98,8 @@ TxReplicator::TxReplicator(
boost::bind(&TxReplicator::commit, this, _1, _2);
dispatch[TxRollbackEvent::KEY] =
boost::bind(&TxReplicator::rollback, this, _1, _2);
+ dispatch[TxMembersEvent::KEY] =
+ boost::bind(&TxReplicator::members, this, _1, _2);
}
TxReplicator::~TxReplicator() {
@@ -130,15 +130,6 @@ void TxReplicator::deliver(const broker:
dm.deliverTo(queue);
}
-void TxReplicator::destroy() {
- {
- sys::Mutex::ScopedLock l(lock);
- if (context.get()) store->abort(*context);
- txBuffer->rollback();
- }
- QueueReplicator::destroy();
-}
-
void TxReplicator::enqueue(const string& data, sys::Mutex::ScopedLock&) {
sys::Mutex::ScopedLock l(lock);
TxEnqueueEvent e;
@@ -209,24 +200,27 @@ void TxReplicator::prepare(const string&
}
}
-void TxReplicator::commit(const string&, sys::Mutex::ScopedLock& l) {
+void TxReplicator::commit(const string&, sys::Mutex::ScopedLock&) {
QPID_LOG(debug, logPrefix << "Commit");
if (context.get()) store->commit(*context);
txBuffer->commit();
- end(l);
}
-void TxReplicator::rollback(const string&, sys::Mutex::ScopedLock& l) {
+void TxReplicator::rollback(const string&, sys::Mutex::ScopedLock&) {
QPID_LOG(debug, logPrefix << "Rollback");
if (context.get()) store->abort(*context);
txBuffer->rollback();
- end(l);
}
-void TxReplicator::end(sys::Mutex::ScopedLock&) {
- // Destroy the tx-queue, which will destroy this via QueueReplicator destroy.
- // FIXME aconway 2013-08-01: what about connection & user ID for destroy?
- haBroker.getBroker().getQueues().destroy(queue->getName());
+void TxReplicator::members(const string& data, sys::Mutex::ScopedLock&) {
+ TxMembersEvent e;
+ decodeStr(data, e);
+ QPID_LOG(debug, logPrefix << "Members: " << e.members);
+ if (!e.members.count(haBroker.getMembership().getSelf())) {
+ QPID_LOG(debug, logPrefix << "Not a member of transaction, terminating");
+ // Destroy the tx-queue, which will destroy this via QueueReplicator destroy.
+ haBroker.deleteQueue(getQueue()->getName());
+ }
}
}} // namespace qpid::ha
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/TxReplicator.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/TxReplicator.h?rev=1510678&r1=1510677&r2=1510678&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/TxReplicator.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/TxReplicator.h Mon Aug 5 19:33:35 2013
@@ -65,7 +65,6 @@ class TxReplicator : public QueueReplica
protected:
void deliver(const broker::Message&);
- void destroy();
private:
@@ -80,13 +79,12 @@ class TxReplicator : public QueueReplica
void prepare(const std::string& data, sys::Mutex::ScopedLock&);
void commit(const std::string& data, sys::Mutex::ScopedLock&);
void rollback(const std::string& data, sys::Mutex::ScopedLock&);
- void end(sys::Mutex::ScopedLock&);
+ void members(const std::string& data, sys::Mutex::ScopedLock&);
std::string logPrefix;
TxEnqueueEvent enq; // Enqueue data for next deliver.
boost::shared_ptr<broker::TxBuffer> txBuffer;
broker::MessageStore* store;
- boost::shared_ptr<BrokerReplicator> brokerReplicator;
std::auto_ptr<broker::TransactionContext> context;
framing::ChannelId channel; // Channel to send prepare-complete.
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/types.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/types.cpp?rev=1510678&r1=1510677&r2=1510678&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/types.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/types.cpp Mon Aug 5 19:33:35 2013
@@ -108,6 +108,24 @@ std::ostream& operator<<(std::ostream& o
return o << m.queue << "[" << m.position << "]=" << m.replicationId;
}
+void UuidSet::encode(framing::Buffer& b) const {
+ b.putLong(size());
+ for (const_iterator i = begin(); i != end(); ++i)
+ b.putRawData(i->data(), i->size());
+}
+
+void UuidSet::decode(framing::Buffer& b) {
+ size_t n = b.getLong();
+ for ( ; n > 0; --n) {
+ types::Uuid id;
+ b.getRawData(const_cast<unsigned char*>(id.data()), id.size());
+ insert(id);
+ }
+}
+
+size_t UuidSet::encodedSize() const {
+ return sizeof(uint32_t) + size()*16;
+}
}} // namespace qpid::ha
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/types.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/types.h?rev=1510678&r1=1510677&r2=1510678&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/types.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/types.h Mon Aug 5 19:33:35 2013
@@ -116,8 +116,13 @@ extern const char* TRANSACTION_REPLICATO
bool startsWith(const std::string& name, const std::string& prefix);
-/** Define IdSet type, not a typedef so we can overload operator << */
-class UuidSet : public std::set<types::Uuid> {};
+/** Define IdSet type, not a typedef so we can overload operator << and add encoding.*/
+class UuidSet : public std::set<types::Uuid> {
+ public:
+ void encode(framing::Buffer&) const;
+ void decode(framing::Buffer&);
+ size_t encodedSize() const;
+};
std::ostream& operator<<(std::ostream& o, const UuidSet& ids);
Modified: qpid/trunk/qpid/cpp/src/tests/ha_test.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/ha_test.py?rev=1510678&r1=1510677&r2=1510678&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/ha_test.py (original)
+++ qpid/trunk/qpid/cpp/src/tests/ha_test.py Mon Aug 5 19:33:35 2013
@@ -48,6 +48,9 @@ class QmfAgent(object):
address, client_properties={"qpid.ha-admin":1}, **kwargs)
self._agent = BrokerAgent(self._connection)
+ def get_queues(self):
+ return [q.values['name'] for q in self._agent.getAllQueues()]
+
def __getattr__(self, name):
a = getattr(self._agent, name)
return a
@@ -107,7 +110,7 @@ class HaBroker(Broker):
ha_port = ha_port or HaPort(test)
args = copy(args)
args += ["--load-module", BrokerTest.ha_lib,
- "--log-enable=trace+:ha::", # FIXME aconway 2013-07-29: debug
+ "--log-enable=debug+:ha::",
# Non-standard settings for faster tests.
"--link-maintenance-interval=0.1",
# Heartbeat and negotiate time are needed so that a broker wont
@@ -276,8 +279,8 @@ class HaCluster(object):
@s_args: args for specific brokers: s_args[i] for broker i.
"""
self.test = test
- self.args = args
- self.s_args = s_args
+ self.args = copy(args)
+ self.s_args = copy(s_args)
self.kwargs = kwargs
self._ports = [HaPort(test) for i in xrange(n)]
self._set_url()
Modified: qpid/trunk/qpid/cpp/src/tests/ha_tests.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/ha_tests.py?rev=1510678&r1=1510677&r2=1510678&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/ha_tests.py (original)
+++ qpid/trunk/qpid/cpp/src/tests/ha_tests.py Mon Aug 5 19:33:35 2013
@@ -20,7 +20,7 @@
import os, signal, sys, time, imp, re, subprocess, glob, random, logging, shutil, math, unittest
import traceback
-from qpid.messaging import Message, SessionError, NotFound, ConnectionError, ReceiverError, Connection, Timeout, Disposition, REJECTED, Empty
+from qpid.messaging import Message, SessionError, NotFound, ConnectionError, ReceiverError, Connection, Timeout, Disposition, REJECTED, Empty, ServerError
from qpid.datatypes import uuid4, UUID
from brokertest import *
from ha_test import *
@@ -1278,7 +1278,7 @@ class StoreTests(BrokerTest):
cluster[0].assert_browse("q1", ["x","y","z"])
cluster[1].assert_browse_backup("q1", ["x","y","z"])
- sn = cluster[0].connect(heartbeat=1).session() # FIXME aconway 2012-09-25: should fail over!
+ sn = cluster[0].connect(heartbeat=1).session()
sn.sender("ex/k1").send("boo")
cluster[0].assert_browse_backup("q1", ["x","y","z", "boo"])
cluster[1].assert_browse_backup("q1", ["x","y","z", "boo"])
@@ -1382,18 +1382,48 @@ class TransactionTests(BrokerTest):
self.assertEqual(open_read(cluster[1].store_log), expect)
def test_tx_backup_fail(self):
- # FIXME aconway 2013-07-31: check exception types, reduce timeout.
cluster = HaCluster(
self, 2, test_store=True, s_args=[[],["--test-store-throw=bang"]])
c = cluster[0].connect()
tx = c.session(transactional=True)
s = tx.sender("q;{create:always,node:{durable:true}}")
for m in ["foo","bang","bar"]: s.send(Message(m, durable=True))
- self.assertRaises(Exception, tx.commit)
+ self.assertRaises(ServerError, tx.commit)
for b in cluster: b.assert_browse_backup("q", [])
self.assertEqual(open_read(cluster[0].store_log), "<begin tx 1>\n<abort tx=1>\n")
self.assertEqual(open_read(cluster[1].store_log), "<begin tx 1>\n<enqueue q foo tx=1>\n<enqueue q bang tx=1>\n<abort tx=1>\n")
+ def test_tx_join_leave(self):
+ """Test cluster members joining/leaving cluster.
+ Also check that tx-queues are cleaned up at end of transaction."""
+
+ def tx_queues(broker):
+ return [q for q in broker.agent().get_queues() if q.startswith("qpid.ha-tx")]
+
+ cluster = HaCluster(self, 3)
+
+ # Leaving
+ tx = cluster[0].connect().session(transactional=True)
+ s = tx.sender("q;{create:always}")
+ s.send("a", sync=True)
+ self.assertEqual([1,1,1], [len(tx_queues(b)) for b in cluster])
+ cluster[1].kill(final=False)
+ s.send("b")
+ self.assertRaises(ServerError, tx.commit)
+ self.assertEqual([[],[]], [tx_queues(b) for b in [cluster[0],cluster[2]]])
+
+ # Joining
+ tx = cluster[0].connect().session(transactional=True)
+ s = tx.sender("q;{create:always}")
+ s.send("foo")
+ tx_q = tx_queues(cluster[0])[0]
+ cluster.restart(1)
+ # Verify the new member should not be in the transaction.
+ # but should receive the result of the transaction via normal replication.
+ cluster[1].wait_no_queue(tx_q)
+ tx.commit()
+ for b in cluster: b.assert_browse_backup("q", ["foo"])
+
if __name__ == "__main__":
outdir = "ha_tests.tmp"
shutil.rmtree(outdir, True)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org