Rebased the wmf branch to the trunk.

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/qmf-devel0.7@916887 13f79535-47bb-0310-9956-ffa450edef68

7 files changed, 146 insertions, 110 deletions

diff --git a/qpid/cpp/src/tests/Makefile.am b/qpid/cpp/src/tests/Makefile.am
index 4d65803ac1..4a931ab680 100644
--- a/qpid/cpp/src/tests/Makefile.am
+++ b/qpid/cpp/src/tests/Makefile.am
@@ -298,7 +298,6 @@ TESTS_ENVIRONMENT = \
     VALGRIND=$(VALGRIND) \
     LIBTOOL="$(LIBTOOL)" \
     QPID_DATA_DIR= \
-    BOOST_TEST_SHOW_PROGRESS=yes \
     $(srcdir)/run_test 
 
 system_tests = client_test quick_perftest quick_topictest run_header_test quick_txtest
diff --git a/qpid/cpp/src/tests/TxMocks.h b/qpid/cpp/src/tests/TxMocks.h
index a34d864bae..72cb50cd21 100644
--- a/qpid/cpp/src/tests/TxMocks.h
+++ b/qpid/cpp/src/tests/TxMocks.h
@@ -23,6 +23,7 @@
 
 
 #include "qpid/Exception.h"
+#include "qpid/Msg.h"
 #include "qpid/broker/TransactionalStore.h"
 #include "qpid/broker/TxOp.h"
 #include <iostream>
diff --git a/qpid/cpp/src/tests/cluster_tests.py b/qpid/cpp/src/tests/cluster_tests.py
index b3274b1b1e..22b7c8f5b8 100755
--- a/qpid/cpp/src/tests/cluster_tests.py
+++ b/qpid/cpp/src/tests/cluster_tests.py
@@ -305,21 +305,47 @@ class StoreTests(BrokerTest):
         self.assertRaises(Exception, lambda: a.ready())
         self.assertRaises(Exception, lambda: b.ready())
 
-    def test_total_failure(self):
-        # Verify we abort with sutiable error message if no clean stores.
-        cluster = self.cluster(0, args=self.args()+["--cluster-size=2"])
-        a = cluster.start("a", expect=EXPECT_EXIT_FAIL, wait=False)
-        b = cluster.start("b", expect=EXPECT_EXIT_FAIL, wait=True)
-        a.kill()
-        b.kill()
-        a = cluster.start("a", expect=EXPECT_EXIT_OK, wait=False)
-        b = cluster.start("b", expect=EXPECT_EXIT_OK, wait=False)
-        self.assertRaises(Exception, lambda: a.ready())
-        self.assertRaises(Exception, lambda: b.ready())
+    def assert_dirty_store(self, broker):
+        self.assertRaises(Exception, lambda: broker.ready())
         msg = re.compile("critical.*no clean store")
-        assert msg.search(readfile(a.log))
-        assert msg.search(readfile(b.log))
+        assert msg.search(readfile(broker.log))
+
+    def test_solo_store_clean(self):
+        # A single node cluster should always leave a clean store.
+        cluster = self.cluster(0, self.args())
+        a = cluster.start("a", expect=EXPECT_EXIT_FAIL)
+        a.send_message("q", Message("x", durable=True))
+        a.kill()
+        a = cluster.start("a")
+        self.assertEqual(a.get_message("q").content, "x")
+
+    def test_last_store_clean(self):
+
+        # Verify that only the last node in a cluster to shut down has
+        # a clean store. Start with cluster of 3, reduce to 1 then
+        # increase again to ensure that a node that was once alone but
+        # finally did not finish as the last node does not get a clean
+        # store.
+        cluster = self.cluster(0, self.args())
+        a = cluster.start("a", expect=EXPECT_EXIT_FAIL)
+        b = cluster.start("b", expect=EXPECT_EXIT_FAIL)
+        c = cluster.start("c", expect=EXPECT_EXIT_FAIL)
+        a.send_message("q", Message("x", durable=True))
+        a.kill()
+        b.kill()          # c is last man
+        time.sleep(0.1)   # pause for c to find out hes last.
+        a = cluster.start("a", expect=EXPECT_EXIT_FAIL) # c no longer last man
+        c.kill()          # a is now last man
+        time.sleep(0.1)   # pause for a to find out hes last.
+        a.kill()          # really last
+        # b & c should be dirty
+        b = cluster.start("b", wait=False, expect=EXPECT_EXIT_OK)
+        self.assert_dirty_store(b)
+        c = cluster.start("c", wait=False, expect=EXPECT_EXIT_OK)
+        self.assert_dirty_store(c)
+        # a should be clean
+        a = cluster.start("a")
+        self.assertEqual(a.get_message("q").content, "x")
 
-        # FIXME aconway 2009-12-03: verify manual restore procedure
 
 
diff --git a/qpid/cpp/src/tests/failover_soak.cpp b/qpid/cpp/src/tests/failover_soak.cpp
index ed5c9cbee5..8bf6eca9e6 100644
--- a/qpid/cpp/src/tests/failover_soak.cpp
+++ b/qpid/cpp/src/tests/failover_soak.cpp
@@ -54,6 +54,8 @@ using namespace qpid::client;
 namespace qpid {
 namespace tests {
 
+vector<pid_t> pids;
+
 typedef vector<ForkedBroker *> brokerVector;
 
 typedef enum
@@ -184,17 +186,29 @@ struct children : public vector<child *>
     int
     checkChildren ( )
     {
-       vector<child *>::iterator i;
-        for ( i = begin(); i != end(); ++ i )
-            if ( (COMPLETED == (*i)->status) && (0 != (*i)->retval) )
-            {
-              cerr << "checkChildren: error on child of type "
-                   << (*i)->type
-                   << endl;
-              return (*i)->retval;
-            }
+      for ( unsigned int i = 0; i < pids.size(); ++ i )
+      {
+        int pid = pids[i];
+        int returned_pid;
+        int status;
 
-        return 0;
+        child * kid = get ( pid );
+
+        if ( kid->status != COMPLETED )
+        {
+          returned_pid = waitpid ( pid, &status, WNOHANG );
+
+          if ( returned_pid == pid )
+          {
+            int exit_status = WEXITSTATUS(status);
+            exited ( pid, exit_status );
+            if ( exit_status )  // this is a child error.
+              return exit_status;
+          }
+        }
+      }
+
+      return 0;
     }
 
 
@@ -323,6 +337,7 @@ startNewBroker ( brokerVector & brokers,
                  int verbosity,
                  int durable )
 {
+        // ("--log-enable=notice+")
     static int brokerId = 0;
     stringstream path, prefix;
     prefix << "soak-" << brokerId;
@@ -516,6 +531,7 @@ startReceivingClient ( brokerVector brokers,
     argv.push_back ( 0 );
 
     pid_t pid = fork();
+    pids.push_back ( pid );
 
     if ( ! pid ) {
         execv ( receiverPath, const_cast<char * const *>(&argv[0]) );
@@ -571,6 +587,7 @@ startSendingClient ( brokerVector brokers,
     argv.push_back ( 0 );
 
     pid_t pid = fork();
+    pids.push_back ( pid );
 
     if ( ! pid ) {
         execv ( senderPath, const_cast<char * const *>(&argv[0]) );
@@ -602,6 +619,7 @@ using namespace qpid::tests;
 int
 main ( int argc, char const ** argv )
 {
+  int brokerKills = 0;
     if ( argc != 11 ) {
         cerr << "Usage: "
              << argv[0]
@@ -625,7 +643,6 @@ main ( int argc, char const ** argv )
     int          n_brokers          = atoi(argv[i++]);
 
     char const * host               = "127.0.0.1";
-    int maxBrokers = 50;
 
     allMyChildren.verbosity = verbosity;
 
@@ -722,104 +739,86 @@ main ( int argc, char const ** argv )
 
 
      int minSleep = 2,
-         maxSleep = 4;
+         maxSleep = 6;
 
+     int totalBrokers = n_brokers;
 
-     for ( int totalBrokers = n_brokers;
-           totalBrokers < maxBrokers;
-           ++ totalBrokers
-         )
+     int loop = 0;
+
+     while ( 1 )
      {
+         ++ loop;
+
+         /*
+         if ( verbosity > 1 )
+           std::cerr << "------- loop " << loop << " --------\n";
+
          if ( verbosity > 0 )
              cout << totalBrokers << " brokers have been added to the cluster.\n\n\n";
+             */
 
          // Sleep for a while. -------------------------
          int sleepyTime = mrand ( minSleep, maxSleep );
-         if ( verbosity > 0 )
-             cout << "Sleeping for " << sleepyTime << " seconds.\n";
          sleep ( sleepyTime );
 
-         // Kill the oldest broker. --------------------------
-         if ( ! killFrontBroker ( brokers, verbosity ) )
+         int bullet = mrand ( 100 );
+         if ( bullet >= 95 )
+	 {
+           fprintf ( stderr, "Killing oldest broker...\n" );
+
+	   // Kill the oldest broker. --------------------------
+	   if ( ! killFrontBroker ( brokers, verbosity ) )
+	   {
+	     allMyChildren.killEverybody();
+	     killAllBrokers ( brokers, 5 );
+	     std::cerr << "END_OF_TEST ERROR_BROKER\n";
+	     return ERROR_KILLING_BROKER;
+	   }
+           ++ brokerKills;
+
+	   // Start a new broker. --------------------------
+	   if ( verbosity > 0 )
+	       cout << "Starting new broker.\n\n";
+
+	   startNewBroker ( brokers,
+			    moduleOrDir,
+			    clusterName,
+			    verbosity,
+			    durable );
+	   ++ totalBrokers;
+           printBrokers ( brokers );
+           cerr << brokerKills << " brokers have been killed.\n\n\n";
+	 }
+
+         int retval = allMyChildren.checkChildren();
+         if ( retval )
          {
-           allMyChildren.killEverybody();
-           killAllBrokers ( brokers, 5 );
-           std::cerr << "END_OF_TEST ERROR_BROKER\n";
-           return ERROR_KILLING_BROKER;
+             std::cerr << "END_OF_TEST ERROR_CLIENT\n";
+             allMyChildren.killEverybody();
+             killAllBrokers ( brokers, 5 );
+             return ERROR_ON_CHILD;
          }
 
-         // Sleep for a while. -------------------------
-         sleepyTime = mrand ( minSleep, maxSleep );
-         if ( verbosity > 1 )
-             cerr << "Sleeping for " << sleepyTime << " seconds.\n";
-         sleep ( sleepyTime );
-
-         // Start a new broker. --------------------------
-         if ( verbosity > 0 )
-             cout << "Starting new broker.\n\n";
-
-         startNewBroker ( brokers,
-                          moduleOrDir,
-                          clusterName,
-                          verbosity,
-                          durable );
-
-         if ( verbosity > 1 )
-             printBrokers ( brokers );
-
          // If all children have exited, quit.
          int unfinished = allMyChildren.unfinished();
-         if ( ! unfinished ) {
+         if ( unfinished == 0 ) {
              killAllBrokers ( brokers, 5 );
 
              if ( verbosity > 1 )
                  cout << "failoverSoak: all children have exited.\n";
-           int retval = allMyChildren.checkChildren();
-           if ( verbosity > 1 )
-             std::cerr << "failoverSoak: checkChildren: " << retval << endl;
-
-           if ( retval )
-           {
-               std::cerr << "END_OF_TEST ERROR_CLIENT\n";
-               return ERROR_ON_CHILD;
-           }
-           else
-           {
-               std::cerr << "END_OF_TEST SUCCESSFUL\n";
-               return HUNKY_DORY;
-           }
-         }
 
-         // Even if some are still running, if there's an error, quit.
-         if ( allMyChildren.checkChildren() )
-         {
-             if ( verbosity > 0 )
-                 cout << "failoverSoak: error on child.\n";
-             allMyChildren.killEverybody();
-             killAllBrokers ( brokers, 5 );
-             std::cerr << "END_OF_TEST ERROR_CLIENT\n";
-             return ERROR_ON_CHILD;
+             std::cerr << "END_OF_TEST SUCCESSFUL\n";
+             return HUNKY_DORY;
          }
 
-         if ( verbosity > 1 ) {
-           std::cerr << "------- next kill-broker loop --------\n";
-           allMyChildren.print();
-         }
      }
 
-     retval = allMyChildren.checkChildren();
-     if ( verbosity > 1 )
-       std::cerr << "failoverSoak: checkChildren: " << retval << endl;
-
-     if ( verbosity > 1 )
-         cout << "failoverSoak: maxBrokers reached.\n";
-
      allMyChildren.killEverybody();
      killAllBrokers ( brokers, 5 );
 
      std::cerr << "END_OF_TEST SUCCESSFUL\n";
 
-     return retval ? ERROR_ON_CHILD : HUNKY_DORY;
+     return HUNKY_DORY;
 }
 
 
diff --git a/qpid/cpp/src/tests/run_failover_soak b/qpid/cpp/src/tests/run_failover_soak
index 69551a51c2..c276e9cc2f 100755
--- a/qpid/cpp/src/tests/run_failover_soak
+++ b/qpid/cpp/src/tests/run_failover_soak
@@ -26,12 +26,12 @@ host=127.0.0.1
 
 unset QPID_NO_MODULE_DIR	# failover_soak uses --module-dir, dont want clash
 MODULES=${MODULES:-$moduledir}
-MESSAGES=${MESSAGES:-1000000}
+MESSAGES=${MESSAGES:-500000}
 REPORT_FREQUENCY=${REPORT_FREQUENCY:-20000}
 VERBOSITY=${VERBOSITY:-10}
 DURABILITY=${DURABILITY:-0}
 N_QUEUES=${N_QUEUES:-1}
-N_BROKERS=${N_BROKERS:-3}
+N_BROKERS=${N_BROKERS:-4}
 
 rm -f soak-*.log
 exec ./failover_soak $MODULES ./declare_queues ./replaying_sender ./resuming_receiver $MESSAGES $REPORT_FREQUENCY $VERBOSITY $DURABILITY $N_QUEUES $N_BROKERS
diff --git a/qpid/cpp/src/tests/test_env.sh.in b/qpid/cpp/src/tests/test_env.sh.in
index 87fbbd128b..07bd4b2bee 100644
--- a/qpid/cpp/src/tests/test_env.sh.in
+++ b/qpid/cpp/src/tests/test_env.sh.in
@@ -73,3 +73,6 @@ exportmodule XML_LIB xml.so
 export QPID_NO_MODULE_DIR=1	# Don't accidentally load installed modules
 export QPID_DATA_DIR= 		# Default to no data dir, not ~/.qpidd
 
+# Options for boost test framework
+export BOOST_TEST_SHOW_PROGRESS=yes
+export BOOST_TEST_CATCH_SYSTEM_ERRORS=no
diff --git a/qpid/cpp/src/tests/verify_cluster_objects b/qpid/cpp/src/tests/verify_cluster_objects
index cea875662f..664b88cb3b 100644..100755
--- a/qpid/cpp/src/tests/verify_cluster_objects
+++ b/qpid/cpp/src/tests/verify_cluster_objects
@@ -75,6 +75,12 @@ class IpAddr:
                 bestAddr = addrPort
         return bestAddr
 
+class ObjectId:
+    """Object identity, use for dictionaries by object id"""
+    def __init__(self, object): self.object = object
+    def __eq__(self, other): return self.object is other.object
+    def __hash__(self): return hash(id(self.object))
+
 class Broker(object):
     def __init__(self, qmf, broker):
         self.broker = broker
@@ -94,6 +100,7 @@ class Broker(object):
             self.uptime = 0
         self.tablesByName = {}
         self.package = "org.apache.qpid.broker"
+        self.id_cache = {}              # Cache for getAbstractId
 
     def getUrl(self):
         return self.broker.getUrl()
@@ -114,13 +121,14 @@ class Broker(object):
     #
     def getAbstractId(self, object):
       """ return a string the of the hierarchical name """
+      if (ObjectId(object) in self.id_cache): return self.id_cache[ObjectId(object)]
       global _debug_recursion
       result = u""
       valstr = u""
       _debug_recursion += 1
       debug_prefix = _debug_recursion
       if (_verbose > 9):
-          print debug_prefix, "  enter gai: props ", self._properties
+          print debug_prefix, "  enter gai: props ", object._properties
       for property, value in object._properties:
 
           # we want to recurse on things which are refs.  we tell by
@@ -138,6 +146,7 @@ class Broker(object):
 
           if property.name == "systemRef":
               _debug_recursion -= 1
+              self.id_cache[ObjectId(object)] = ""
               return ""
 
           if property.index:
@@ -176,6 +185,7 @@ class Broker(object):
           if (_verbose > 9):
               print debug_prefix, "    id ", self, " -> ", result
       _debug_recursion -= 1
+      self.id_cache[ObjectId(object)] = result
       return result
 
     def loadTable(self, cls):
@@ -196,13 +206,12 @@ class Broker(object):
             # error (ie, the name-generation code is busted) if we do
             key = self.getAbstractId(obj)
             if key in self.tablesByName[cls.getClassName()]:
-                print "internal error: collision for %s on key %s\n" % (obj, key)
-                sys.exit(1)
+                raise Exception("internal error: collision for %s on key %s\n"
+                                % (obj, key))
                 
-            self.tablesByName[cls.getClassName()][self.getAbstractId(obj)] = obj
-#            sys.exit(1)
+            self.tablesByName[cls.getClassName()][key] = obj
             if _verbose > 1:
-                print "   ", obj.getObjectId(), " ", obj.getIndex(), " ", self.getAbstractId(obj)
+                print "   ", obj.getObjectId(), " ", obj.getIndex(), " ", key
 
 
 class BrokerManager:
@@ -253,9 +262,10 @@ class BrokerManager:
                     raise Exception("Invalid URL 2")
                 addrList.append((tokens[1], tokens[2]))
 
-            # Find the address in the list that is most likely to be in the same subnet as the address
-            # with which we made the original QMF connection.  This increases the probability that we will
-            # be able to reach the cluster member.
+            # Find the address in the list that is most likely to be
+            # in the same subnet as the address with which we made the
+            # original QMF connection.  This increases the probability
+            # that we will be able to reach the cluster member.
 
             best = hostAddr.bestAddr(addrList)
             bestUrl = best[0] + ":" + best[1]
@@ -284,8 +294,7 @@ class BrokerManager:
                 if _verbose > 0:
                     print "   ", b
         else:
-            print "Failed - Not a cluster"
-            sys.exit(1)
+            raise Exception("Failed - Not a cluster")
 
         failures = []
 
@@ -348,11 +357,10 @@ class BrokerManager:
             print "Failures:"
             for failure in failures:
                 print "  %s" % failure
-            sys.exit(1)
+            raise Exception("Failures")
 
         if _verbose > 0:
             print "Success"
-        sys.exit(0)
 
 ##
 ## Main Program