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 2011/03/07 22:01:49 UTC
svn commit: r1078947 - in /qpid/trunk/qpid/cpp/src:
qpid/broker/Connection.cpp qpid/broker/SemanticState.cpp
qpid/sys/ClusterSafe.cpp qpid/sys/ClusterSafe.h tests/brokertest.py
tests/cluster_test_logs.py tests/cluster_tests.py
Author: aconway
Date: Mon Mar 7 21:01:49 2011
New Revision: 1078947
URL: http://svn.apache.org/viewvc?rev=1078947&view=rev
Log:
QPID-3121: Cluster management inconsistency when using persistent store.
With the store doing async completions, completion IO callbacks could
be queued differently on different nodes. This led to inconsistent
management changes in a cluster when a connection was modified in an
IO callback.
Fix was to mark IO callback processing as not cluster safe, so
connections don't record management stats during an IO callback.
Test changes:
- enable durable tests in test_management.
- add substitutions to mask known issue of inconsistent "stats changed" messages.
- add transactional client to test_management.
- ignore heartbeat connection close logs in cluster_test_logs.py
- make brokertest.retry more accurate
- fix minor bug in brokertest.log_ready.
Modified:
qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp
qpid/trunk/qpid/cpp/src/qpid/sys/ClusterSafe.cpp
qpid/trunk/qpid/cpp/src/qpid/sys/ClusterSafe.h
qpid/trunk/qpid/cpp/src/tests/brokertest.py
qpid/trunk/qpid/cpp/src/tests/cluster_test_logs.py
qpid/trunk/qpid/cpp/src/tests/cluster_tests.py
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp?rev=1078947&r1=1078946&r2=1078947&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp Mon Mar 7 21:01:49 2011
@@ -340,6 +340,9 @@ void Connection::closed(){ // Physically
void Connection::doIoCallbacks() {
{
ScopedLock<Mutex> l(ioCallbackLock);
+ // Although IO callbacks execute in the connection thread context, they are
+ // not cluster safe because they are queued for execution in non-IO threads.
+ ClusterUnsafeScope cus;
while (!ioCallbacks.empty()) {
boost::function0<void> cb = ioCallbacks.front();
ioCallbacks.pop();
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp?rev=1078947&r1=1078946&r2=1078947&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp Mon Mar 7 21:01:49 2011
@@ -88,7 +88,7 @@ void SemanticState::closed() {
//prevent requeued messages being redelivered to consumers
for (ConsumerImplMap::iterator i = consumers.begin(); i != consumers.end(); i++) {
disable(i->second);
- }
+ }
if (dtxBuffer.get()) {
dtxBuffer->fail();
}
@@ -107,7 +107,7 @@ bool SemanticState::exists(const string&
return consumers.find(consumerTag) != consumers.end();
}
-void SemanticState::consume(const string& tag,
+void SemanticState::consume(const string& tag,
Queue::shared_ptr queue, bool ackRequired, bool acquire,
bool exclusive, const string& resumeId, uint64_t resumeTtl, const FieldTable& arguments)
{
@@ -197,7 +197,7 @@ void SemanticState::endDtx(const std::st
dtxBuffer->fail();
} else {
dtxBuffer->markEnded();
- }
+ }
dtxBuffer.reset();
}
@@ -257,9 +257,9 @@ void SemanticState::record(const Deliver
const std::string QPID_SYNC_FREQUENCY("qpid.sync_frequency");
-SemanticState::ConsumerImpl::ConsumerImpl(SemanticState* _parent,
- const string& _name,
- Queue::shared_ptr _queue,
+SemanticState::ConsumerImpl::ConsumerImpl(SemanticState* _parent,
+ const string& _name,
+ Queue::shared_ptr _queue,
bool ack,
bool _acquire,
bool _exclusive,
@@ -268,20 +268,20 @@ SemanticState::ConsumerImpl::ConsumerImp
const framing::FieldTable& _arguments
-) :
+) :
Consumer(_acquire),
- parent(_parent),
- name(_name),
- queue(_queue),
- ackExpected(ack),
+ parent(_parent),
+ name(_name),
+ queue(_queue),
+ ackExpected(ack),
acquire(_acquire),
- blocked(true),
+ blocked(true),
windowing(true),
exclusive(_exclusive),
resumeId(_resumeId),
resumeTtl(_resumeTtl),
arguments(_arguments),
- msgCredit(0),
+ msgCredit(0),
byteCredit(0),
notifyEnabled(true),
syncFrequency(_arguments.getAsInt(QPID_SYNC_FREQUENCY)),
@@ -292,7 +292,7 @@ SemanticState::ConsumerImpl::ConsumerImp
{
ManagementAgent* agent = parent->session.getBroker().getManagementAgent();
qpid::management::Manageable* ms = dynamic_cast<qpid::management::Manageable*> (&(parent->session));
-
+
if (agent != 0)
{
mgmtObject = new _qmf::Subscription(agent, this, ms , queue->GetManagementObject()->getObjectId() ,name,
@@ -334,7 +334,7 @@ bool SemanticState::ConsumerImpl::delive
if (!ackExpected && acquire) record.setEnded();//allows message to be released now its been delivered
if (windowing || ackExpected || !acquire) {
parent->record(record);
- }
+ }
if (acquire && !ackExpected) {
queue->dequeue(0, msg);
}
@@ -354,7 +354,7 @@ bool SemanticState::ConsumerImpl::accept
// checkCredit fails because the message is to big, we should
// remain on queue's listener list for possible smaller messages
// in future.
- //
+ //
blocked = !(filter(msg) && checkCredit(msg));
return !blocked;
}
@@ -375,7 +375,7 @@ void SemanticState::ConsumerImpl::alloca
{
assertClusterSafe();
uint32_t originalMsgCredit = msgCredit;
- uint32_t originalByteCredit = byteCredit;
+ uint32_t originalByteCredit = byteCredit;
if (msgCredit != 0xFFFFFFFF) {
msgCredit--;
}
@@ -385,7 +385,7 @@ void SemanticState::ConsumerImpl::alloca
QPID_LOG(debug, "Credit allocated for " << ConsumerName(*this)
<< ", was " << " bytes: " << originalByteCredit << " msgs: " << originalMsgCredit
<< " now bytes: " << byteCredit << " msgs: " << msgCredit);
-
+
}
bool SemanticState::ConsumerImpl::checkCredit(intrusive_ptr<Message>& msg)
@@ -399,7 +399,7 @@ bool SemanticState::ConsumerImpl::checkC
return enoughCredit;
}
-SemanticState::ConsumerImpl::~ConsumerImpl()
+SemanticState::ConsumerImpl::~ConsumerImpl()
{
if (mgmtObject != 0)
mgmtObject->resourceDestroy ();
@@ -417,7 +417,7 @@ void SemanticState::unsubscribe(Consumer
Queue::shared_ptr queue = c->getQueue();
if(queue) {
queue->cancel(c);
- if (queue->canAutoDelete() && !queue->hasExclusiveOwner()) {
+ if (queue->canAutoDelete() && !queue->hasExclusiveOwner()) {
Queue::tryAutoDelete(session.getBroker(), queue);
}
}
@@ -460,7 +460,7 @@ const std::string nullstring;
void SemanticState::route(intrusive_ptr<Message> msg, Deliverable& strategy) {
msg->setTimestamp(getSession().getBroker().getExpiryPolicy());
-
+
std::string exchangeName = msg->getExchangeName();
if (!cacheExchange || cacheExchange->getName() != exchangeName)
cacheExchange = session.getBroker().getExchanges().get(exchangeName);
@@ -469,7 +469,7 @@ void SemanticState::route(intrusive_ptr<
/* verify the userid if specified: */
std::string id =
msg->hasProperties<MessageProperties>() ? msg->getProperties<MessageProperties>()->getUserId() : nullstring;
-
+
if (authMsg && !id.empty() && !(id == userID || (isDefaultRealm && id == userName)))
{
QPID_LOG(debug, "authorised user id : " << userID << " but user id in message declared as " << id);
@@ -487,7 +487,7 @@ void SemanticState::route(intrusive_ptr<
if (!strategy.delivered) {
//TODO:if discard-unroutable, just drop it
- //TODO:else if accept-mode is explicit, reject it
+ //TODO:else if accept-mode is explicit, reject it
//else route it to alternate exchange
if (cacheExchange->getAlternate()) {
cacheExchange->getAlternate()->route(strategy, msg->getRoutingKey(), msg->getApplicationHeaders());
@@ -516,7 +516,7 @@ void SemanticState::ConsumerImpl::reques
}
bool SemanticState::complete(DeliveryRecord& delivery)
-{
+{
ConsumerImplMap::iterator i = consumers.find(delivery.getTag());
if (i != consumers.end()) {
i->second->complete(delivery);
@@ -544,7 +544,7 @@ void SemanticState::recover(bool requeue
unacked.clear();
for_each(copy.rbegin(), copy.rend(), mem_fun_ref(&DeliveryRecord::requeue));
}else{
- for_each(unacked.begin(), unacked.end(), boost::bind(&DeliveryRecord::redeliver, _1, this));
+ for_each(unacked.begin(), unacked.end(), boost::bind(&DeliveryRecord::redeliver, _1, this));
//unconfirmed messages re redelivered and therefore have their
//id adjusted, confirmed messages are not and so the ordering
//w.r.t id is lost
@@ -676,7 +676,7 @@ Queue::shared_ptr SemanticState::getQueu
}
AckRange SemanticState::findRange(DeliveryId first, DeliveryId last)
-{
+{
return DeliveryRecord::findRange(unacked, first, last);
}
@@ -767,13 +767,13 @@ void SemanticState::accepted(const Seque
//in transactional mode, don't dequeue or remove, just
//maintain set of acknowledged messages:
accumulatedAck.add(commands);
-
+
if (dtxBuffer.get()) {
//if enlisted in a dtx, copy the relevant slice from
//unacked and record it against that transaction
TxOp::shared_ptr txAck(new DtxAck(accumulatedAck, unacked));
accumulatedAck.clear();
- dtxBuffer->enlist(txAck);
+ dtxBuffer->enlist(txAck);
//mark the relevant messages as 'ended' in unacked
//if the messages are already completed, they can be
@@ -795,7 +795,6 @@ void SemanticState::accepted(const Seque
}
void SemanticState::completed(const SequenceSet& commands) {
- assertClusterSafe();
DeliveryRecords::iterator removed =
remove_if(unacked.begin(), unacked.end(),
isInSequenceSetAnd(commands,
@@ -806,7 +805,6 @@ void SemanticState::completed(const Sequ
void SemanticState::attached()
{
- assertClusterSafe();
for (ConsumerImplMap::iterator i = consumers.begin(); i != consumers.end(); i++) {
i->second->enableNotify();
session.getConnection().outputTasks.addOutputTask(i->second.get());
@@ -816,7 +814,6 @@ void SemanticState::attached()
void SemanticState::detached()
{
- assertClusterSafe();
for (ConsumerImplMap::iterator i = consumers.begin(); i != consumers.end(); i++) {
i->second->disableNotify();
session.getConnection().outputTasks.removeOutputTask(i->second.get());
Modified: qpid/trunk/qpid/cpp/src/qpid/sys/ClusterSafe.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/ClusterSafe.cpp?rev=1078947&r1=1078946&r2=1078947&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/ClusterSafe.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/ClusterSafe.cpp Mon Mar 7 21:01:49 2011
@@ -51,6 +51,16 @@ ClusterSafeScope::~ClusterSafeScope() {
inContext = save;
}
+ClusterUnsafeScope::ClusterUnsafeScope() {
+ save = inContext;
+ inContext = false;
+}
+
+ClusterUnsafeScope::~ClusterUnsafeScope() {
+ assert(!inContext);
+ inContext = save;
+}
+
void enableClusterSafe() { inCluster = true; }
}} // namespace qpid::sys
Modified: qpid/trunk/qpid/cpp/src/qpid/sys/ClusterSafe.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/ClusterSafe.h?rev=1078947&r1=1078946&r2=1078947&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/ClusterSafe.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/ClusterSafe.h Mon Mar 7 21:01:49 2011
@@ -53,10 +53,8 @@ QPID_COMMON_EXTERN void assertClusterSaf
QPID_COMMON_EXTERN bool isClusterSafe();
/**
- * Base class for classes that encapsulate state which is replicated
- * to all members of a cluster. Acts as a marker for clustered state
- * and provides functions to assist detecting bugs in cluster
- * behavior.
+ * Mark a scope as cluster safe. Sets isClusterSafe in constructor and resets
+ * to previous value in destructor.
*/
class ClusterSafeScope {
public:
@@ -67,6 +65,18 @@ class ClusterSafeScope {
};
/**
+ * Mark a scope as cluster unsafe. Clears isClusterSafe in constructor and resets
+ * to previous value in destructor.
+ */
+class ClusterUnsafeScope {
+ public:
+ ClusterUnsafeScope();
+ ~ClusterUnsafeScope();
+ private:
+ bool save;
+};
+
+/**
* Enable cluster-safe assertions. By default they are no-ops.
* Called by cluster code.
*/
Modified: qpid/trunk/qpid/cpp/src/tests/brokertest.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/brokertest.py?rev=1078947&r1=1078946&r2=1078947&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/brokertest.py (original)
+++ qpid/trunk/qpid/cpp/src/tests/brokertest.py Mon Mar 7 21:01:49 2011
@@ -67,7 +67,7 @@ class ExceptionWrapper:
def __init__(self, obj, msg):
self.obj = obj
self.msg = msg
-
+
def __getattr__(self, name):
func = getattr(self.obj, name)
if type(func) != callable:
@@ -97,11 +97,12 @@ def retry(function, timeout=10, delay=.0
"""Call function until it returns True or timeout expires.
Double the delay for each retry. Return True if function
returns true, False if timeout expires."""
+ deadline = time.time() + timeout
while not function():
- if delay > timeout: delay = timeout
+ remaining = deadline - time.time()
+ if remaining <= 0: return False
+ delay = min(delay, remaining)
time.sleep(delay)
- timeout -= delay
- if timeout <= 0: return False
delay *= 2
return True
@@ -191,7 +192,7 @@ class Popen(subprocess.Popen):
def unexpected(self,msg):
err = error_line(self.outfile("err")) or error_line(self.outfile("out"))
raise BadProcessStatus("%s %s%s" % (self.pname, msg, err))
-
+
def stop(self): # Clean up at end of test.
try:
if self.expect == EXPECT_UNKNOWN:
@@ -213,7 +214,7 @@ class Popen(subprocess.Popen):
self.unexpected("expected error")
finally:
self.wait() # Clean up the process.
-
+
def communicate(self, input=None):
if input:
self.stdin.write(input)
@@ -231,7 +232,7 @@ class Popen(subprocess.Popen):
def poll(self, _deadstate=None): # _deadstate required by base class in python 2.4
if self.returncode is None:
# Pass _deadstate only if it has been set, there is no _deadstate
- # parameter in Python 2.6
+ # parameter in Python 2.6
if _deadstate is None: ret = subprocess.Popen.poll(self)
else: ret = subprocess.Popen.poll(self, _deadstate)
@@ -255,7 +256,7 @@ class Popen(subprocess.Popen):
os.kill( self.pid , signal.SIGTERM)
except AttributeError: # no os.kill, using taskkill.. (Windows only)
os.popen('TASKKILL /PID ' +str(self.pid) + ' /F')
-
+
def kill(self):
try: subprocess.Popen.kill(self)
except AttributeError: # No terminate method
@@ -289,7 +290,7 @@ class Broker(Popen):
while (os.path.exists(self.log)):
self.log = "%s-%d.log" % (self.name, i)
i += 1
-
+
def get_log(self):
return os.path.abspath(self.log)
@@ -319,7 +320,7 @@ class Broker(Popen):
cmd += ["--log-to-file", self.log]
cmd += ["--log-to-stderr=no"]
if log_level != None:
- cmd += ["--log-enable=%s" % log_level]
+ cmd += ["--log-enable=%s" % log_level]
self.datadir = self.name
cmd += ["--data-dir", self.datadir]
Popen.__init__(self, cmd, expect, drain=False)
@@ -362,7 +363,7 @@ class Broker(Popen):
s = c.session(str(qpid.datatypes.uuid4()))
s.queue_declare(queue=queue)
c.close()
-
+
def _prep_sender(self, queue, durable, xprops):
s = queue + "; {create:always, node:{durable:" + str(durable)
if xprops != None: s += ", x-declare:{" + xprops + "}"
@@ -406,13 +407,14 @@ class Broker(Popen):
def log_ready(self):
"""Return true if the log file exists and contains a broker ready message"""
- if self._log_ready: return True
- self._log_ready = find_in_file("notice Broker running", self.log)
+ if not self._log_ready:
+ self._log_ready = find_in_file("notice Broker running", self.log)
+ return self._log_ready
def ready(self, **kwargs):
"""Wait till broker is ready to serve clients"""
# First make sure the broker is listening by checking the log.
- if not retry(self.log_ready, timeout=30):
+ if not retry(self.log_ready, timeout=60):
raise Exception(
"Timed out waiting for broker %s%s"%(self.name, error_line(self.log,5)))
# Create a connection and a session. For a cluster broker this will
@@ -421,8 +423,8 @@ class Broker(Popen):
c = self.connect(**kwargs)
try: c.session()
finally: c.close()
- except: raise RethrownException(
- "Broker %s failed ready test%s"%(self.name,error_line(self.log, 5)))
+ except Exception,e: raise RethrownException(
+ "Broker %s not responding: (%s)%s"%(self.name,e,error_line(self.log, 5)))
def store_state(self):
uuids = open(os.path.join(self.datadir, "cluster", "store.status")).readlines()
@@ -431,7 +433,7 @@ class Broker(Popen):
if uuids[0] == null_uuid: return "empty"
if uuids[1] == null_uuid: return "dirty"
return "clean"
-
+
class Cluster:
"""A cluster of brokers in a test."""
@@ -486,7 +488,7 @@ class BrokerTest(TestCase):
rootdir = os.getcwd()
def configure(self, config): self.config=config
-
+
def setUp(self):
outdir = self.config.defines.get("OUTDIR") or "brokertest.tmp"
self.dir = os.path.join(self.rootdir, outdir, self.id())
@@ -561,7 +563,7 @@ class StoppableThread(Thread):
self.stopped = True
self.join()
if self.error: raise self.error
-
+
class NumberedSender(Thread):
"""
Thread to run a sender client and send numbered messages until stopped.
@@ -620,7 +622,7 @@ class NumberedSender(Thread):
self.join()
self.write_message(-1) # end-of-messages marker.
if self.error: raise self.error
-
+
class NumberedReceiver(Thread):
"""
Thread to run a receiver client and verify it receives
@@ -647,7 +649,7 @@ class NumberedReceiver(Thread):
def read_message(self):
return int(self.receiver.stdout.readline())
-
+
def run(self):
try:
self.received = 0
@@ -679,7 +681,7 @@ class ErrorGenerator(StoppableThread):
self.broker=broker
broker.test.cleanup_stop(self)
self.start()
-
+
def run(self):
c = self.broker.connect_old()
try:
Modified: qpid/trunk/qpid/cpp/src/tests/cluster_test_logs.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/cluster_test_logs.py?rev=1078947&r1=1078946&r2=1078947&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/cluster_test_logs.py (original)
+++ qpid/trunk/qpid/cpp/src/tests/cluster_test_logs.py Mon Mar 7 21:01:49 2011
@@ -61,10 +61,10 @@ def filter_log(log):
'warning CLOSING .* unsent data',
'Inter-broker link ',
'Running in a cluster, marking store',
- 'debug Sending keepalive signal to watchdog',
- 'last broker standing joined by 1 replicas, updating queue policies.'
+ 'debug Sending keepalive signal to watchdog', # Watchdog timer thread
+ 'last broker standing joined by 1 replicas, updating queue policies.',
+ 'Connection .* timed out: closing' # heartbeat connection close
])
- skip_re = re.compile(skip)
# Regex to match a UUID
uuid='\w\w\w\w\w\w\w\w-\w\w\w\w-\w\w\w\w-\w\w\w\w-\w\w\w\w\w\w\w\w\w\w\w\w'
# Substitutions to remove expected differences
@@ -82,6 +82,13 @@ def filter_log(log):
(r' map={.*_object_name:([^,}]*)[,}].*', r' \1'), # V2 map - just keep name
(r'\d+-\d+-\d+--\d+', 'X-X-X--X'), # V1 Object IDs
]
+ # Substitutions to mask known issue: durable test shows inconsistent "changed stats for com.redhat.rhm.store:journal" messages.
+ skip += '|Changed V[12] statistics com.redhat.rhm.store:journal'
+ subs += [(r'to=console.obj.1.0.com.redhat.rhm.store.journal props=\d+ stats=\d+',
+ 'to=console.obj.1.0.com.redhat.rhm.store.journal props=NN stats=NN')]
+
+ skip_re = re.compile(skip)
+ subs = [(re.compile(pattern), subst) for pattern, subst in subs]
for l in open(log):
if skip_re.search(l): continue
for pattern,subst in subs: l = re.sub(pattern,subst,l)
Modified: qpid/trunk/qpid/cpp/src/tests/cluster_tests.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/cluster_tests.py?rev=1078947&r1=1078946&r2=1078947&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/cluster_tests.py (original)
+++ qpid/trunk/qpid/cpp/src/tests/cluster_tests.py Mon Mar 7 21:01:49 2011
@@ -574,8 +574,10 @@ class LongTests(BrokerTest):
"""Start ordinary clients for a broker."""
cmds=[
["qpid-tool", "localhost:%s"%(broker.port())],
- ["qpid-perftest", "--count", 50000,
+ ["qpid-perftest", "--count=5000", "--durable=yes",
"--base-name", str(qpid.datatypes.uuid4()), "--port", broker.port()],
+ ["qpid-txtest", "--queue-base-name", "tx-%s"%str(qpid.datatypes.uuid4()),
+ "--port", broker.port()],
["qpid-queue-stats", "-a", "localhost:%s" %(broker.port())],
["testagent", "localhost", str(broker.port())] ]
clients.append([ClientLoop(broker, cmd) for cmd in cmds])
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org