Author: aconway
Date: Wed Aug 6 14:17:19 2008
New Revision: 683416
URL: http://svn.apache.org/viewvc?rev=683416&view=rev
Log:
- Added OutputTask::hasOutput() test.
- Cluster only sends doOutput events when hasOutput()
Added:
incubator/qpid/trunk/qpid/cpp/src/tests/restart_cluster (with props)
Modified:
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.h
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ConnectionInterceptor.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AggregateOutput.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AggregateOutput.h
incubator/qpid/trunk/qpid/cpp/src/qpid/sys/OutputTask.h
incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am
incubator/qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp?rev=683416&r1=683415&r2=683416&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp Wed Aug 6
14:17:19 2008
@@ -193,6 +193,8 @@
}
}
+bool Connection::hasOutput() { return outputTasks.hasOutput(); }
+
bool Connection::doOutput() { return doOutputFn(); }
bool Connection::doOutputImpl() {
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.h?rev=683416&r1=683415&r2=683416&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.h Wed Aug 6
14:17:19 2008
@@ -76,6 +76,7 @@
void received(framing::AMQFrame& frame);
void idleOut();
void idleIn();
+ bool hasOutput();
bool doOutput();
void closed();
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp?rev=683416&r1=683415&r2=683416&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp Wed Aug 6 14:17:19
2008
@@ -212,6 +212,11 @@
}
}
+bool Queue::empty() const {
+ Mutex::ScopedLock locker(messageLock);
+ return messages.empty();
+}
+
bool Queue::consumeNextMessage(QueuedMessage& m, Consumer& c)
{
while (true) {
@@ -348,7 +353,6 @@
}
}
consumerCount++;
-
if (mgmtObject != 0)
mgmtObject->inc_consumerCount ();
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h?rev=683416&r1=683415&r2=683416&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h Wed Aug 6 14:17:19
2008
@@ -107,7 +107,6 @@
void notify();
void removeListener(Consumer&);
- void addListener(Consumer&);
bool isExcluded(boost::intrusive_ptr<Message>& msg);
@@ -115,6 +114,9 @@
void popAndDequeue();
public:
+ // FIXME aconway 2008-08-06: was private, verify if needed public.
+ void addListener(Consumer&);
+
virtual void notifyDurableIOComplete();
typedef boost::shared_ptr<Queue> shared_ptr;
@@ -126,6 +128,8 @@
management::Manageable* parent = 0);
~Queue();
+ bool empty() const;
+
bool dispatch(Consumer&);
void create(const qpid::framing::FieldTable& settings);
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp?rev=683416&r1=683415&r2=683416&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp Wed Aug 6
14:17:19 2008
@@ -590,6 +590,11 @@
unacked.erase(range.start, range.end);
}
+bool SemanticState::ConsumerImpl::hasOutput() {
+ queue->addListener(*this);
+ return !queue->empty();
+}
+
bool SemanticState::ConsumerImpl::doOutput()
{
//TODO: think through properly
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.h?rev=683416&r1=683415&r2=683416&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.h Wed Aug 6
14:17:19 2008
@@ -96,6 +96,7 @@
Queue::shared_ptr getQueue() { return queue; }
bool isBlocked() const { return blocked; }
+ bool hasOutput();
bool doOutput();
};
@@ -180,6 +181,7 @@
void release(DeliveryId first, DeliveryId last, bool setRedelivered);
void reject(DeliveryId first, DeliveryId last);
void handle(boost::intrusive_ptr<Message> msg);
+ bool hasOutput() { return outputTasks.hasOutput(); }
bool doOutput() { return outputTasks.doOutput(); }
//final 0-10 spec (completed and accepted are distinct):
Modified:
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ConnectionInterceptor.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ConnectionInterceptor.cpp?rev=683416&r1=683415&r2=683416&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ConnectionInterceptor.cpp
(original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ConnectionInterceptor.cpp
Wed Aug 6 14:17:19 2008
@@ -82,15 +82,16 @@
}
bool ConnectionInterceptor::doOutput() {
- cluster.send(AMQFrame(in_place<ClusterConnectionDoOutputBody>()), this);
+ if (connection->hasOutput()) {
+ printf("doOutput send %p\n", (void*)this);
+ cluster.send(AMQFrame(in_place<ClusterConnectionDoOutputBody>()),
this);
+ }
+
return false;
}
void ConnectionInterceptor::deliverDoOutput() {
- // FIXME aconway 2008-07-16: review thread safety.
- // All connection processing happens in cluster queue, only read & write
- // (from mutex-locked frameQueue) happens in reader/writer threads.
- //
+ printf("doOutput deliver %p\n", (void*)this);
doOutputNext();
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AggregateOutput.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AggregateOutput.cpp?rev=683416&r1=683415&r2=683416&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AggregateOutput.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AggregateOutput.cpp Wed Aug 6
14:17:19 2008
@@ -31,6 +31,12 @@
control.activateOutput();
}
+bool AggregateOutput::hasOutput() {
+ for (TaskList::const_iterator i = tasks.begin(); i != tasks.end(); ++i)
+ if ((*i)->hasOutput()) return true;
+ return false;
+}
+
bool AggregateOutput::doOutput()
{
bool result = false;
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AggregateOutput.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AggregateOutput.h?rev=683416&r1=683415&r2=683416&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AggregateOutput.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AggregateOutput.h Wed Aug 6
14:17:19 2008
@@ -43,6 +43,7 @@
void activateOutput();
//all the following will be called on the same thread
bool doOutput();
+ bool hasOutput();
void addOutputTask(OutputTask* t);
void removeOutputTask(OutputTask* t);
};
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/OutputTask.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/OutputTask.h?rev=683416&r1=683415&r2=683416&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/OutputTask.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/OutputTask.h Wed Aug 6 14:17:19
2008
@@ -28,7 +28,17 @@
{
public:
virtual ~OutputTask() {}
+ /** Generate some output.
+ [EMAIL PROTECTED] true if output was generated, false if there is no
work to do.
+ */
virtual bool doOutput() = 0;
+
+ /** Check if there may be work to do, but don't do it.
+ * @return True if there may be work to do, false if there is none.
+ * Can to return a false positive, to allow implementations to do a
+ * faster check than doOutput(). Must never return a false negative.
+ */
+ virtual bool hasOutput() = 0;
};
}
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am?rev=683416&r1=683415&r2=683416&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am Wed Aug 6 14:17:19 2008
@@ -135,7 +135,8 @@
MessageUtils.h \
TestMessageStore.h \
MockConnectionInputHandler.h \
- TxMocks.h
+ TxMocks.h \
+ start_cluster stop_cluster restart_cluster
check_LTLIBRARIES += libdlclose_noop.la
libdlclose_noop_la_LDFLAGS = -module -rpath $(abs_builddir)
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp?rev=683416&r1=683415&r2=683416&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp Wed Aug 6
14:17:19 2008
@@ -162,6 +162,8 @@
}
};
+#if 0 // FIXME aconway 2008-08-06:
+
QPID_AUTO_TEST_CASE(CpgBasic) {
// Verify basic functionality of cpg. This will catch any
// openais configuration or permission errors.
@@ -182,7 +184,6 @@
BOOST_CHECK_EQUAL(0, cb.configChanges[1]);
}
-
QPID_AUTO_TEST_CASE(testForkedBroker) {
// Verify the ForkedBroker works as expected.
const char* argv[] = { "", "--auth=no", "--no-data-dir",
"--log-prefix=testForkedBroker" };
@@ -249,7 +250,7 @@
BOOST_CHECK_EQUAL(0u, c1.session.queueQuery("q").getMessageCount());
BOOST_CHECK_EQUAL(0u, c2.session.queueQuery("q").getMessageCount());
}
-
+#endif
QPID_AUTO_TEST_CASE(testDequeueWaitingSubscription) {
ClusterFixture cluster(3);
// First start a subscription.
Added: incubator/qpid/trunk/qpid/cpp/src/tests/restart_cluster
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/restart_cluster?rev=683416&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/restart_cluster (added)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/restart_cluster Wed Aug 6 14:17:19
2008
@@ -0,0 +1,18 @@
+#!/bin/sh
+# Re-start a cluster on the local host.
+
+srcdir=`dirname $0`
+$srcdir/stop_cluster
+exec $srcdir/start_cluster "$@"
+#!/bin/sh
+# Re-start a cluster on the local host.
+
+srcdir=`dirname $0`
+$srcdir/stop_cluster
+exec $srcdir/start_cluster "$@"
+#!/bin/sh
+# Re-start a cluster on the local host.
+
+srcdir=`dirname $0`
+$srcdir/stop_cluster
+exec $srcdir/start_cluster "$@"
Propchange: incubator/qpid/trunk/qpid/cpp/src/tests/restart_cluster
------------------------------------------------------------------------------
svn:executable = *
