Added cluster option --cluster-size.

--cluster-size=N means that during start-up the cluster waits to have N members
before accepting any clients.


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

8 files changed, 77 insertions, 30 deletions

diff --git a/cpp/src/qpid/cluster/Cluster.cpp b/cpp/src/qpid/cluster/Cluster.cpp
index 9756ad0a62..5e962e9767 100644
--- a/cpp/src/qpid/cluster/Cluster.cpp
+++ b/cpp/src/qpid/cluster/Cluster.cpp
@@ -208,7 +208,7 @@ Cluster::Cluster(const ClusterSettings& set, broker::Broker& b) :
     decoder(boost::bind(&Cluster::deliverFrame, this, _1)),
     discarding(true),
     state(INIT),
-    initMap(self),
+    initMap(self, settings.size),
     lastSize(0),
     lastBroker(false),
     updateRetracted(false),
@@ -403,8 +403,7 @@ void Cluster::flagError(
                  << ": " << msg);
         leave(l);
     }
-    else if (settings.checkErrors)
-        error.error(connection, type, map.getFrameSeq(), map.getMembers(), msg);
+    error.error(connection, type, map.getFrameSeq(), map.getMembers(), msg);
 }
 
 // Handler for deliverFrameQueue.
@@ -423,7 +422,7 @@ void Cluster::deliveredFrame(const EventFrame& efConst) {
         deliverEventQueue.start(); 
     }
     // Process each frame through the error checker.
-    if (settings.checkErrors && error.isUnresolved()) {
+    if (error.isUnresolved()) {
         error.delivered(e);
         while (error.canProcess())  // There is a frame ready to process.
             processFrame(error.getNext(), l);
@@ -874,7 +873,7 @@ std::ostream& operator<<(std::ostream& o, const Cluster& cluster) {
     };
     assert(sizeof(STATE)/sizeof(*STATE) == Cluster::LEFT+1);
     o << "cluster(" << cluster.self << " " << STATE[cluster.state];
-    if (cluster.settings.checkErrors && cluster.error.isUnresolved()) o << "/error";
+    if (cluster.error.isUnresolved()) o << "/error";
     return o << ")";;
 }
 
diff --git a/cpp/src/qpid/cluster/ClusterPlugin.cpp b/cpp/src/qpid/cluster/ClusterPlugin.cpp
index 289ec672a8..4eec388866 100644
--- a/cpp/src/qpid/cluster/ClusterPlugin.cpp
+++ b/cpp/src/qpid/cluster/ClusterPlugin.cpp
@@ -71,9 +71,8 @@ struct ClusterOptions : public Options {
 #if HAVE_LIBCMAN_H
             ("cluster-cman", optValue(settings.quorum), "Integrate with Cluster Manager (CMAN) cluster.")
 #endif
+            ("cluster-size", optValue(settings.size, "N"), "Wait for N cluster members before allowing clients to connect.")
             ("cluster-read-max", optValue(settings.readMax,"N"), "Experimental: flow-control limit  reads per connection. 0=no limit.")
-            // TODO aconway 2009-05-20: temporary, remove
-            ("cluster-check-errors", optValue(settings.checkErrors, "yes|no"), "Enable/disable cluster error checks. Normally should be enabled.")
             ;
     }
 };
diff --git a/cpp/src/qpid/cluster/ClusterSettings.h b/cpp/src/qpid/cluster/ClusterSettings.h
index 8820e8a5ac..8e708aa139 100644
--- a/cpp/src/qpid/cluster/ClusterSettings.h
+++ b/cpp/src/qpid/cluster/ClusterSettings.h
@@ -34,10 +34,9 @@ struct ClusterSettings {
     bool quorum;
     size_t readMax;
     std::string username, password, mechanism;
-    bool checkErrors;
+    size_t size;
 
-    ClusterSettings() : quorum(false), readMax(10),
-                        checkErrors(true) // TODO aconway 2009-05-20: remove this option.
+    ClusterSettings() : quorum(false), readMax(10), size(1)
     {}
   
     Url getUrl(uint16_t port) const {
diff --git a/cpp/src/qpid/cluster/InitialStatusMap.cpp b/cpp/src/qpid/cluster/InitialStatusMap.cpp
index 6d27b3ae72..f2251f4043 100644
--- a/cpp/src/qpid/cluster/InitialStatusMap.cpp
+++ b/cpp/src/qpid/cluster/InitialStatusMap.cpp
@@ -28,8 +28,8 @@ using namespace boost;
 namespace qpid {
 namespace cluster {
 
-InitialStatusMap::InitialStatusMap(const MemberId& self_)
-    : self(self_), completed(), resendNeeded()
+InitialStatusMap::InitialStatusMap(const MemberId& self_, size_t size_)
+    : self(self_), completed(), resendNeeded(), size(size_)
 {}
 
 void InitialStatusMap::configChange(const MemberSet& members) {
@@ -83,7 +83,8 @@ bool InitialStatusMap::isActive(const Map::value_type& v) {
 }
 
 bool InitialStatusMap::isComplete() {
-    return !map.empty() && find_if(map.begin(), map.end(), &notInitialized) == map.end();
+    return !map.empty() && find_if(map.begin(), map.end(), &notInitialized) == map.end()
+        && (map.size() >= size);
 }
 
 bool InitialStatusMap::transitionToComplete() {
diff --git a/cpp/src/qpid/cluster/InitialStatusMap.h b/cpp/src/qpid/cluster/InitialStatusMap.h
index 4605a4c1fe..9e9b71e363 100644
--- a/cpp/src/qpid/cluster/InitialStatusMap.h
+++ b/cpp/src/qpid/cluster/InitialStatusMap.h
@@ -37,7 +37,7 @@ class InitialStatusMap
   public:
     typedef framing::ClusterInitialStatusBody Status;
 
-    InitialStatusMap(const MemberId& self);
+    InitialStatusMap(const MemberId& self, size_t size);
     /** Process a config change. @return true if we need to re-send our status */
     void configChange(const MemberSet& newConfig);
     /** @return true if we need to re-send status */
@@ -71,6 +71,7 @@ class InitialStatusMap
     MemberSet firstConfig;
     MemberId self;
     bool completed, resendNeeded;
+    size_t size;
 };
 }} // namespace qpid::cluster
 
diff --git a/cpp/src/tests/InitialStatusMap.cpp b/cpp/src/tests/InitialStatusMap.cpp
index 70b077b695..c3587965e5 100644
--- a/cpp/src/tests/InitialStatusMap.cpp
+++ b/cpp/src/tests/InitialStatusMap.cpp
@@ -40,7 +40,7 @@ Status newcomerStatus(const Uuid& id=Uuid()) { return Status(ProtocolVersion(),
 
 QPID_AUTO_TEST_CASE(testFirstInCluster) {
     // Single member is first in cluster.
-    InitialStatusMap map(MemberId(0));
+    InitialStatusMap map(MemberId(0), 1);
     Uuid id(true);
     BOOST_CHECK(!map.isComplete());
     MemberSet members = list_of(MemberId(0));
@@ -56,7 +56,7 @@ QPID_AUTO_TEST_CASE(testFirstInCluster) {
 
 QPID_AUTO_TEST_CASE(testJoinExistingCluster) {
     // Single member 0 joins existing cluster 1,2
-    InitialStatusMap map(MemberId(0));
+    InitialStatusMap map(MemberId(0), 1);
     Uuid id(true);
     MemberSet members = list_of(MemberId(0))(MemberId(1))(MemberId(2));
     map.configChange(members);
@@ -79,7 +79,7 @@ QPID_AUTO_TEST_CASE(testJoinExistingCluster) {
 
 QPID_AUTO_TEST_CASE(testMultipleFirstInCluster) {
     // Multiple members 0,1,2 join at same time.
-    InitialStatusMap map(MemberId(1)); // self is 1
+    InitialStatusMap map(MemberId(1), 1); // self is 1
     Uuid id(true);
     MemberSet members = list_of(MemberId(0))(MemberId(1))(MemberId(2));
     map.configChange(members);
@@ -99,7 +99,7 @@ QPID_AUTO_TEST_CASE(testMultipleFirstInCluster) {
 
 QPID_AUTO_TEST_CASE(testMultipleJoinExisting) {
     // Multiple members 1,2,3 join existing cluster containing 0.
-    InitialStatusMap map(MemberId(2)); // self is 2
+    InitialStatusMap map(MemberId(2), 1); // self is 2
     Uuid id(true);
     MemberSet members = list_of(MemberId(0))(MemberId(1))(MemberId(2))(MemberId(3));
     map.configChange(members);
@@ -119,7 +119,7 @@ QPID_AUTO_TEST_CASE(testMultipleJoinExisting) {
 
 QPID_AUTO_TEST_CASE(testMembersLeave) {
     // Test that map completes if members leave rather than send status.
-    InitialStatusMap map(MemberId(0));
+    InitialStatusMap map(MemberId(0), 1);
     Uuid id(true);
     map.configChange(list_of(MemberId(0))(MemberId(1))(MemberId(2)));
     map.received(MemberId(0), newcomerStatus());
@@ -134,7 +134,7 @@ QPID_AUTO_TEST_CASE(testMembersLeave) {
 
 QPID_AUTO_TEST_CASE(testInteveningConfig) {
     // Multiple config changes arrives before we complete the map.
-    InitialStatusMap map(MemberId(0));
+    InitialStatusMap map(MemberId(0), 1);
     Uuid id(true);
 
     map.configChange(list_of<MemberId>(0)(1));
@@ -159,6 +159,20 @@ QPID_AUTO_TEST_CASE(testInteveningConfig) {
     BOOST_CHECK_EQUAL(map.getClusterId(), id);
 }
 
+QPID_AUTO_TEST_CASE(testInitialSize) {
+    InitialStatusMap map(MemberId(0), 3);
+    map.configChange(list_of<MemberId>(0)(1));
+    map.received(MemberId(0), newcomerStatus());
+    map.received(MemberId(1), newcomerStatus());
+    BOOST_CHECK(!map.isComplete());
+
+    map.configChange(list_of<MemberId>(0)(1)(2));
+    map.received(MemberId(0), newcomerStatus());
+    map.received(MemberId(1), newcomerStatus());
+    map.received(MemberId(2), newcomerStatus());
+    BOOST_CHECK(map.isComplete());
+}
+
 QPID_AUTO_TEST_SUITE_END()
 
 }} // namespace qpid::tests
diff --git a/cpp/src/tests/cluster_tests.py b/cpp/src/tests/cluster_tests.py
index 3ded6c103e..ed39277f77 100755
--- a/cpp/src/tests/cluster_tests.py
+++ b/cpp/src/tests/cluster_tests.py
@@ -26,13 +26,8 @@ from qpid.messaging import Message
 from threading import Thread
 
 
-class ClusterTests(BrokerTest):
-    """Cluster tests with support for testing with a store plugin."""
-
-    def duration(self):
-        d = self.config.defines.get("DURATION")
-        if d: return float(d)*60
-        else: return 3
+class ShortTests(BrokerTest):
+    """Short cluster functionality tests."""
 
     def test_message_replication(self):
         """Test basic cluster message replication."""
@@ -57,6 +52,42 @@ class ClusterTests(BrokerTest):
         self.assertEqual("y", m.content)
         s2.connection.close()
 
+    def test_cluster_size(self):
+        """Verify cluster startup waits for N brokers if --cluster-size=N"""
+        class ConnectThread(Thread):
+            def __init__(self, broker):
+                Thread.__init__(self)
+                self.broker=broker
+                self.connected = False
+                self.error = None
+
+            def run(self):
+                try:
+                    self.broker.connect()
+                    self.connected = True
+                except Exception, e: self.error = RethrownException(e)
+
+        cluster = self.cluster(1, args=["--cluster-size=3"], wait_for_start=False)
+        c = ConnectThread(cluster[0])
+        c.start()
+        time.sleep(.01)
+        assert not c.connected
+        cluster.start(wait_for_start=False)
+        time.sleep(.01)
+        assert not c.connected
+        cluster.start(wait_for_start=False)
+        c.join(1)
+        assert not c.isAlive()       # Join didn't time out
+        assert c.connected
+        if c.error: raise c.error
+
+class LongTests(BrokerTest):
+    """Tests that can run for a long time if -DDURATION=<minutes> is set"""
+    def duration(self):
+        d = self.config.defines.get("DURATION")
+        if d: return float(d)*60
+        else: return 3                  # Default is to be quick
+
     def test_failover(self):
         """Test fail-over during continuous send-receive with errors"""
 
@@ -84,7 +115,8 @@ class ClusterTests(BrokerTest):
         receiver.stop(sender.sent)
         for i in range(i, len(cluster)): cluster[i].kill()
 
-class ClusterStoreTests(BrokerTest):
+
+class StoreTests(BrokerTest):
     """
     Cluster tests that can only be run if there is a store available.
     """
diff --git a/cpp/src/tests/run_cluster_tests b/cpp/src/tests/run_cluster_tests
index 014233d8d3..b090f71852 100755
--- a/cpp/src/tests/run_cluster_tests
+++ b/cpp/src/tests/run_cluster_tests
@@ -45,8 +45,10 @@ rm -rf $OUTDIR
 mkdir -p $OUTDIR
 
 # Ignore tests requiring a store by default. 
-TESTS="-i cluster_tests.ClusterStoreTests.* -I $srcdir/cluster_tests.fail $*"
+CLUSTER_TESTS_IGNORE=${CLUSTER_TESTS_IGNORE:--i cluster_tests.StoreTests.* -I $srcdir/cluster_tests.fail}
+CLUSTER_TESTS=${CLUSTER_TESTS:-$*}
 
-with_ais_group $PYTHON_COMMANDS/qpid-python-test -DOUTDIR=$OUTDIR -m cluster_tests $TESTS || exit 1
+set -x
+with_ais_group $PYTHON_COMMANDS/qpid-python-test -DOUTDIR=$OUTDIR -m cluster_tests $CLUSTER_TESTS_IGNORE $CLUSTER_TESTS || exit 1
 rm -rf $OUTDIR
 #exit 0