synchronized with trunk except for ruby dir

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

1 files changed, 874 insertions, 215 deletions

diff --git a/cpp/src/qpid/cluster/Cluster.cpp b/cpp/src/qpid/cluster/Cluster.cpp
index 05ab9148b5..d049001eb0 100644
--- a/cpp/src/qpid/cluster/Cluster.cpp
+++ b/cpp/src/qpid/cluster/Cluster.cpp
@@ -16,305 +16,964 @@
  *
  */
 
-#include "Cluster.h"
-#include "ConnectionInterceptor.h"
-
+/** CLUSTER IMPLEMENTATION OVERVIEW
+ *
+ * The cluster works on the principle that if all members of the
+ * cluster receive identical input, they will all produce identical
+ * results. cluster::Connections intercept data received from clients
+ * and multicast it via CPG. The data is processed (passed to the
+ * broker::Connection) only when it is received from CPG in cluster
+ * order. Each cluster member has Connection objects for directly
+ * connected clients and "shadow" Connection objects for connections
+ * to other members.
+ *
+ * This assumes that all broker actions occur deterministically in
+ * response to data arriving on client connections. There are two
+ * situations where this assumption fails:
+ *  - sending data in response to polling local connections for writabiliy.
+ *  - taking actions based on a timer or timestamp comparison.
+ *
+ * IMPORTANT NOTE: any time code is added to the broker that uses timers,
+ * the cluster may need to be updated to take account of this.
+ * 
+ *
+ * USE OF TIMESTAMPS IN THE BROKER
+ *  
+ * The following are the current areas where broker uses timers or timestamps:
+ * 
+ * - Producer flow control: broker::SemanticState uses connection::getClusterOrderOutput.
+ *   a FrameHandler that sends frames to the client via the cluster. Used by broker::SessionState
+ *   
+ * - QueueCleaner, Message TTL: uses ExpiryPolicy, which is implemented by cluster::ExpiryPolicy.
+ * 
+ * - Connection heartbeat: sends connection controls, not part of session command counting so OK to ignore.
+ * 
+ * - LinkRegistry: only cluster elder is ever active for links.
+ * 
+ * - management::ManagementBroker: uses MessageHandler supplied by  cluster
+ *   to send messages to the broker via the cluster.
+ *   
+ * - Dtx: not yet supported with cluster.  
+ *
+ * cluster::ExpiryPolicy implements the strategy for message expiry.
+ *
+ * CLUSTER PROTOCOL OVERVIEW
+ * 
+ * Messages sent to/from CPG are called Events.
+ *
+ * An Event carries a ConnectionId, which includes a MemberId and a
+ * connection number.
+ * 
+ * Events are either
+ *  - Connection events: non-0 connection number and are associated with a connection.
+ *  - Cluster Events: 0 connection number, are not associated with a connection.
+ * 
+ * Events are further categorized as:
+ *  - Control: carries method frame(s) that affect cluster behavior.
+ *  - Data: carries raw data received from a client connection.
+ *
+ * The cluster defines extensions to the AMQP command set in ../../../xml/cluster.xml
+ * which defines two classes:
+ *  - cluster: cluster control information.
+ *  - cluster.connection: control information for a specific connection.
+ *
+ * The following combinations are legal:
+ *  - Data frames carrying connection data.
+ *  - Cluster control events carrying cluster commands.
+ *  - Connection control events carrying cluster.connection commands.
+ *  - Connection control events carrying non-cluster frames: frames sent to the client.
+ *    e.g. flow-control frames generated on a timer.
+ *
+ * CLUSTER INITIALIZATION OVERVIEW
+ *
+ * When a new member joins the CPG group, all members (including the
+ * new one) multicast their "initial status." The new member is in
+ * INIT mode until it gets a complete set of initial status messages
+ * from all cluster members.
+ *
+ * The newcomer uses initial status to determine
+ *  - The cluster UUID
+ *  - Am I speaking the correct version of the cluster protocol?
+ *  - Do I need to get an update from an existing active member?
+ *  - Can I recover from my own store?
+ *
+ * Initialization happens in the Cluster constructor (plugin
+ * early-init phase) because it needs to be done before the store
+ * initializes. In INIT mode sending & receiving from the cluster are
+ * done single-threaded, bypassing the normal PollableQueues because
+ * the Poller is not active at this point to service them.
+ */
+#include "qpid/Exception.h"
+#include "qpid/cluster/Cluster.h"
+#include "qpid/cluster/ClusterSettings.h"
+#include "qpid/cluster/Connection.h"
+#include "qpid/cluster/UpdateClient.h"
+#include "qpid/cluster/RetractClient.h"
+#include "qpid/cluster/FailoverExchange.h"
+#include "qpid/cluster/UpdateExchange.h"
+
+#include "qpid/assert.h"
+#include "qmf/org/apache/qpid/cluster/ArgsClusterStopClusterNode.h"
+#include "qmf/org/apache/qpid/cluster/Package.h"
 #include "qpid/broker/Broker.h"
-#include "qpid/broker/SessionState.h"
 #include "qpid/broker/Connection.h"
+#include "qpid/broker/NullMessageStore.h"
+#include "qpid/broker/QueueRegistry.h"
+#include "qpid/broker/SessionState.h"
+#include "qpid/broker/SignalHandler.h"
 #include "qpid/framing/AMQFrame.h"
-#include "qpid/framing/ClusterNotifyBody.h"
-#include "qpid/framing/ClusterConnectionCloseBody.h"
+#include "qpid/framing/AMQP_AllOperations.h"
+#include "qpid/framing/AllInvoker.h"
+#include "qpid/framing/ClusterConfigChangeBody.h"
+#include "qpid/framing/ClusterConnectionDeliverCloseBody.h"
+#include "qpid/framing/ClusterConnectionAbortBody.h"
+#include "qpid/framing/ClusterRetractOfferBody.h"
+#include "qpid/framing/ClusterConnectionDeliverDoOutputBody.h"
+#include "qpid/framing/ClusterReadyBody.h"
+#include "qpid/framing/ClusterShutdownBody.h"
+#include "qpid/framing/ClusterUpdateOfferBody.h"
+#include "qpid/framing/ClusterUpdateRequestBody.h"
+#include "qpid/framing/ClusterConnectionAnnounceBody.h"
+#include "qpid/framing/ClusterErrorCheckBody.h"
+#include "qpid/framing/MessageTransferBody.h"
+#include "qpid/log/Helpers.h"
 #include "qpid/log/Statement.h"
+#include "qpid/management/IdAllocator.h"
+#include "qpid/management/ManagementAgent.h"
 #include "qpid/memory.h"
-#include "qpid/shared_ptr.h"
+#include "qpid/sys/Thread.h"
 
+#include <boost/shared_ptr.hpp>
 #include <boost/bind.hpp>
 #include <boost/cast.hpp>
+#include <boost/current_function.hpp>
 #include <algorithm>
 #include <iterator>
 #include <map>
+#include <ostream>
 
 namespace qpid {
 namespace cluster {
+using namespace qpid;
 using namespace qpid::framing;
 using namespace qpid::sys;
+using namespace qpid::cluster;
+using namespace framing::cluster;
 using namespace std;
-using broker::Connection;
+using management::ManagementAgent;
+using management::ManagementObject;
+using management::Manageable;
+using management::Args;
+namespace _qmf = ::qmf::org::apache::qpid::cluster;
+
+/**
+ * NOTE: must increment this number whenever any incompatible changes in
+ * cluster protocol/behavior are made. It allows early detection and
+ * sensible reporting of an attempt to mix different versions in a
+ * cluster.
+ *
+ * Currently use SVN revision to avoid clashes with versions from
+ * different branches.
+ */
+const uint32_t Cluster::CLUSTER_VERSION = 884125;
 
-ostream& operator <<(ostream& out, const Cluster& cluster) {
-    return out << cluster.name.str() << "-" << cluster.self;
-}
+struct ClusterDispatcher : public framing::AMQP_AllOperations::ClusterHandler {
+    qpid::cluster::Cluster& cluster;
+    MemberId member;
+    Cluster::Lock& l;
+    ClusterDispatcher(Cluster& c, const MemberId& id, Cluster::Lock& l_) : cluster(c), member(id), l(l_) {}
 
-ostream& operator<<(ostream& out, const Cluster::MemberMap::value_type& m) {
-    return out << m.first << "=" << m.second.url;
-}
+    void updateRequest(const std::string& url) { cluster.updateRequest(member, url, l); }
 
-ostream& operator <<(ostream& out, const Cluster::MemberMap& members) {
-    ostream_iterator<Cluster::MemberMap::value_type> o(out, " ");
-    copy(members.begin(), members.end(), o);
-    return out;
-}
+    void initialStatus(uint32_t version, bool active, const Uuid& clusterId,
+                       uint8_t storeState, const Uuid& shutdownId)
+    {
+        cluster.initialStatus(member, version, active, clusterId, 
+                              framing::cluster::StoreState(storeState), shutdownId, l);
+    }
+    void ready(const std::string& url) { cluster.ready(member, url, l); }
+    void configChange(const std::string& current) { cluster.configChange(member, current, l); }
+    void updateOffer(uint64_t updatee) {
+        cluster.updateOffer(member, updatee, l);
+    }
+    void retractOffer(uint64_t updatee) { cluster.retractOffer(member, updatee, l); }
+    void messageExpired(uint64_t id) { cluster.messageExpired(member, id, l); }
+    void errorCheck(uint8_t type, const framing::SequenceNumber& frameSeq) {
+        cluster.errorCheck(member, type, frameSeq, l);
+    }
+
+    void shutdown(const Uuid& id) { cluster.shutdown(member, id, l); }
+
+    bool invoke(AMQBody& body) { return framing::invoke(*this, body).wasHandled(); }
+};
 
-Cluster::Cluster(const std::string& name_, const Url& url_, broker::Broker& b) :
-    broker(&b),
+Cluster::Cluster(const ClusterSettings& set, broker::Broker& b) :
+    settings(set), 
+    broker(b),
+    mgmtObject(0),
     poller(b.getPoller()),
     cpg(*this),
-    name(name_),
-    url(url_),
+    name(settings.name),
+    myUrl(settings.url.empty() ? Url() : Url(settings.url)),
     self(cpg.self()),
-    cpgDispatchHandle(cpg,
-                      boost::bind(&Cluster::dispatch, this, _1), // read
-                      0,                                         // write
-                      boost::bind(&Cluster::disconnect, this, _1) // disconnect
-    ),
-    deliverQueue(boost::bind(&Cluster::deliverQueueCb, this, _1, _2)),
-    mcastQueue(boost::bind(&Cluster::mcastQueueCb, this, _1, _2))
+    clusterId(true),
+    expiryPolicy(new ExpiryPolicy(mcast, self, broker.getTimer())),
+    mcast(cpg, poller, boost::bind(&Cluster::leave, this)),
+    dispatcher(cpg, poller, boost::bind(&Cluster::leave, this)),
+    deliverEventQueue(boost::bind(&Cluster::deliveredEvent, this, _1),
+                      boost::bind(&Cluster::leave, this),
+                      "Error decoding events",
+                      poller),
+    deliverFrameQueue(boost::bind(&Cluster::deliveredFrame, this, _1),
+                      boost::bind(&Cluster::leave, this),
+                      "Error delivering frames",
+                      poller),
+    quorum(boost::bind(&Cluster::leave, this)),
+    decoder(boost::bind(&Cluster::deliverFrame, this, _1)),
+    discarding(true),
+    state(INIT),
+    initMap(self, settings.size),
+    store(broker.getDataDir().getPath()),
+    lastSize(0),
+    lastBroker(false),
+    updateRetracted(false),
+    error(*this)
 {
-    broker->addFinalizer(boost::bind(&Cluster::leave, this));
-    QPID_LOG(trace, "Joining cluster: " << name_);
-    cpg.join(name);
-    notify();
+    mAgent = broker.getManagementAgent();
+    if (mAgent != 0){
+        _qmf::Package  packageInit(mAgent);
+        mgmtObject = new _qmf::Cluster (mAgent, this, &broker,name,myUrl.str());
+        mAgent->addObject (mgmtObject);
+        mgmtObject->set_status("JOINING");
+    }
 
-    // FIXME aconway 2008-08-11: can we remove this loop?
-    // Dispatch till we show up in the cluster map.
-    while (empty()) 
-        cpg.dispatchOne();
+    // Failover exchange provides membership updates to clients.
+    failoverExchange.reset(new FailoverExchange(this));
+    broker.getExchanges().registerExchange(failoverExchange);
+
+    // Update exchange is used during updates to replicate messages
+    // without modifying delivery-properties.exchange.
+    broker.getExchanges().registerExchange(
+        boost::shared_ptr<broker::Exchange>(new UpdateExchange(this)));
+    // Load my store status before we go into initialization
+    if (! broker::NullMessageStore::isNullStore(&broker.getStore())) {
+        store.load();
+        if (store.getClusterId())
+            clusterId = store.getClusterId(); // Use stored ID if there is one.
+        QPID_LOG(notice, "Cluster store state: " << store)
+    }
 
-    // Start dispatching from the poller.
-    cpgDispatchHandle.startWatch(poller);
-    deliverQueue.start(poller);
-    mcastQueue.start(poller);
+    cpg.join(name);
+    // Pump the CPG dispatch manually till we get initialized. 
+    while (state == INIT)
+        cpg.dispatchOne();
 }
 
 Cluster::~Cluster() {
-    for (ShadowConnectionMap::iterator i = shadowConnectionMap.begin();
-         i != shadowConnectionMap.end();
-         ++i)
-    {
-        i->second->dirtyClose(); 
+    if (updateThread.id()) updateThread.join(); // Join the previous updatethread.
+}
+
+void Cluster::initialize() {
+    if (settings.quorum) quorum.start(poller);
+    if (myUrl.empty())
+        myUrl = Url::getIpAddressesUrl(broker.getPort(broker::Broker::TCP_TRANSPORT));
+    // Cluster constructor will leave us in either READY or JOINER state.
+    switch (state) {
+      case READY:
+        mcast.mcastControl(ClusterReadyBody(ProtocolVersion(), myUrl.str()), self);
+        break;
+      case JOINER:
+        mcast.mcastControl(ClusterUpdateRequestBody(ProtocolVersion(), myUrl.str()), self);
+        break;
+      default:
+        assert(0);
     }
-    std::for_each(localConnectionSet.begin(), localConnectionSet.end(), boost::bind(&ConnectionInterceptor::dirtyClose, _1));
+    QPID_LOG(notice, *this << (state == READY ? " joined" : " joining") << " cluster " << name);
+    broker.getKnownBrokers = boost::bind(&Cluster::getUrls, this);
+    broker.setExpiryPolicy(expiryPolicy);
+    dispatcher.start();
+    deliverEventQueue.start();
+    deliverFrameQueue.start();
+
+    // Add finalizer last for exception safety.
+    broker.addFinalizer(boost::bind(&Cluster::brokerShutdown, this)); 
 }
 
-// local connection initializes plugins
-void Cluster::initialize(broker::Connection& c) {
-    bool isLocal = &c.getOutput() != &shadowOut;
-    if (isLocal)
-        localConnectionSet.insert(new ConnectionInterceptor(c, *this));
+// Called in connection thread to insert a client connection.
+void Cluster::addLocalConnection(const boost::intrusive_ptr<Connection>& c) {
+    QPID_LOG(info, *this << " new local connection " << c->getId());
+    localConnections.insert(c);
+    assert(c->getId().getMember() == self);
+    // Announce the connection to the cluster.
+    if (c->isLocalClient())
+        mcast.mcastControl(ClusterConnectionAnnounceBody(ProtocolVersion(),
+                                                         c->getBrokerConnection().getSSF() ),
+                           c->getId());
 }
 
-void Cluster::leave() {
-    Mutex::ScopedLock l(lock);
-    if (!broker) return;                               // Already left.
-    // At this point the poller has already been shut down so
-    // no dispatches can occur thru the cpgDispatchHandle.
-    // 
-    // FIXME aconway 2008-08-11: assert this is the cae.
-    
-    QPID_LOG(debug, "Leaving cluster " << *this);
-    cpg.leave(name);
-    // broker= is set to 0 when the final config-change is delivered.
-    while(broker) {
-        Mutex::ScopedUnlock u(lock);
-        cpg.dispatchAll();
-    }
-    cpg.shutdown();
+// Called in connection thread to insert an updated shadow connection.
+void Cluster::addShadowConnection(const boost::intrusive_ptr<Connection>& c) {
+    QPID_LOG(info, *this << " new shadow connection " << c->getId());
+    // Safe to use connections here because we're pre-catchup, stalled
+    // and discarding, so deliveredFrame is not processing any
+    // connection events.
+    assert(discarding);         
+    pair<ConnectionMap::iterator, bool> ib
+        = connections.insert(ConnectionMap::value_type(c->getId(), c));
+    assert(ib.second);
+}
+
+void Cluster::erase(const ConnectionId& id) {
+    Lock l(lock);    
+    erase(id,l);
 }
 
-template <class T> void decodePtr(Buffer& buf, T*& ptr) {
-    uint64_t value = buf.getLongLong();
-    ptr = reinterpret_cast<T*>(value);
+// Called by Connection::deliverClose() in deliverFrameQueue thread.
+void Cluster::erase(const ConnectionId& id, Lock&) {
+    QPID_LOG(info, *this << " connection closed " << id);
+    connections.erase(id);
+    decoder.erase(id);
 }
 
-template <class T> void encodePtr(Buffer& buf, T* ptr) {
-    uint64_t value = reinterpret_cast<uint64_t>(ptr);
-    buf.putLongLong(value);
+std::vector<string> Cluster::getIds() const {
+    Lock l(lock);
+    return getIds(l);
 }
 
-void Cluster::send(const AMQFrame& frame, ConnectionInterceptor* connection) {
-    QPID_LOG(trace, "MCAST [" << connection << "] " << frame);
-    mcastQueue.push(Message(frame, self, connection));
+std::vector<string> Cluster::getIds(Lock&) const {
+    return map.memberIds();
 }
 
-void Cluster::mcastQueueCb(const MessageQueue::iterator& begin,
-                           const MessageQueue::iterator& end)
+std::vector<Url> Cluster::getUrls() const {
+    Lock l(lock);
+    return getUrls(l);
+}
+
+std::vector<Url> Cluster::getUrls(Lock&) const {
+    return map.memberUrls();
+} 
+
+void Cluster::leave() { 
+    Lock l(lock);
+    leave(l);
+}
+
+#define LEAVE_TRY(STMT) try { STMT; }                                   \
+    catch (const std::exception& e) {                                   \
+        QPID_LOG(warning, *this << " error leaving cluster: " << e.what()); \
+    } do {} while(0)
+
+void Cluster::leave(Lock&) { 
+    if (state != LEFT) {
+        state = LEFT;
+        QPID_LOG(notice, *this << " leaving cluster " << name);
+        // Finalize connections now now to avoid problems later in destructor.
+        LEAVE_TRY(localConnections.clear());
+        LEAVE_TRY(connections.clear());
+        LEAVE_TRY(broker::SignalHandler::shutdown());
+    }
+}
+
+// Deliver CPG message.
+void Cluster::deliver(
+    cpg_handle_t /*handle*/,
+    const cpg_name* /*group*/,
+    uint32_t nodeid,
+    uint32_t pid,
+    void* msg,
+    int msg_len) 
 {
-    // Static is OK because there is only one cluster allowed per
-    // process and only one thread in mcastQueueCb at a time.
-    static char buffer[64*1024]; // FIXME aconway 2008-07-04: buffer management.
-    MessageQueue::iterator i = begin;
-    while (i != end) {
-        Buffer buf(buffer, sizeof(buffer));
-        while (i != end && buf.available() > i->frame.size() + sizeof(uint64_t)) {
-            i->frame.encode(buf);
-            encodePtr(buf, i->connection);
-            ++i;
+    MemberId from(nodeid, pid);
+    framing::Buffer buf(static_cast<char*>(msg), msg_len);
+    Event e(Event::decodeCopy(from, buf));
+    deliverEvent(e);
+}
+
+void Cluster::deliverEvent(const Event& e) {
+    // During initialization, execute events directly in the same thread.
+    // Once initialized, push to pollable queue to be processed in another thread.
+    if (state == INIT)
+        deliveredEvent(e);
+    else
+        deliverEventQueue.push(e);
+}
+
+void Cluster::deliverFrame(const EventFrame& e) {
+    // During initialization, execute events directly in the same thread.
+    // Once initialized, push to pollable queue to be processed in another thread.
+    if (state == INIT)
+        deliveredFrame(e);
+    else
+        deliverFrameQueue.push(e);
+}
+
+const ClusterUpdateOfferBody* castUpdateOffer(const framing::AMQBody* body) {
+    return  (body && body->getMethod() &&
+             body->getMethod()->isA<ClusterUpdateOfferBody>()) ?
+        static_cast<const ClusterUpdateOfferBody*>(body) : 0;
+}
+
+const ClusterConnectionAnnounceBody* castAnnounce( const framing::AMQBody *body) {
+    return  (body && body->getMethod() &&
+             body->getMethod()->isA<ClusterConnectionAnnounceBody>()) ?
+      static_cast<const ClusterConnectionAnnounceBody*>(body) : 0;
+}
+
+// Handler for deliverEventQueue.
+// This thread decodes frames from events.
+void Cluster::deliveredEvent(const Event& e) {
+    if (e.isCluster()) {
+        EventFrame ef(e, e.getFrame());
+        // Stop the deliverEventQueue on update offers.
+        // This preserves the connection decoder fragments for an update.
+        const ClusterUpdateOfferBody* offer = castUpdateOffer(ef.frame.getBody());
+        if (offer) {
+            QPID_LOG(info, *this << " stall for update offer from " << e.getMemberId()
+                     << " to " << MemberId(offer->getUpdatee()));
+            deliverEventQueue.stop();
+        }
+        deliverFrame(ef);
+    }
+    else if(!discarding) { 
+        if (e.isControl())
+            deliverFrame(EventFrame(e, e.getFrame()));
+        else {
+            try { decoder.decode(e, e.getData()); }
+            catch (const Exception& ex) {
+                // Close a connection that is sending us invalid data.
+                QPID_LOG(error, *this << " aborting connection "
+                         << e.getConnectionId() << ": " << ex.what());
+                framing::AMQFrame abort((ClusterConnectionAbortBody()));
+                deliverFrame(EventFrame(EventHeader(CONTROL, e.getConnectionId()), abort));
+            }
         }
-        iovec iov = { buffer, buf.getPosition() };
-        cpg.mcast(name, &iov, 1);
     }
 }
 
-void Cluster::notify() {
-    send(AMQFrame(in_place<ClusterNotifyBody>(ProtocolVersion(), url.str())), 0);
+void Cluster::flagError(
+    Connection& connection, ErrorCheck::ErrorType type, const std::string& msg)
+{
+    Mutex::ScopedLock l(lock);
+    if (connection.isCatchUp()) {
+        QPID_LOG(critical, *this << " error on update connection " << connection
+                 << ": " << msg);
+        leave(l);
+    }
+    error.error(connection, type, map.getFrameSeq(), map.getMembers(), msg);
 }
 
-size_t Cluster::size() const {
+// Handler for deliverFrameQueue.
+// This thread executes the main logic.
+void Cluster::deliveredFrame(const EventFrame& efConst) {
     Mutex::ScopedLock l(lock);
-    return members.size();
+    if (state == LEFT) return;
+    EventFrame e(efConst);
+    const ClusterUpdateOfferBody* offer = castUpdateOffer(e.frame.getBody());
+    if (offer && error.isUnresolved()) {
+        // We can't honour an update offer that is delivered while an
+        // error is in progress so replace it with a retractOffer and re-start
+        // the event queue.
+        e.frame = AMQFrame(
+            ClusterRetractOfferBody(ProtocolVersion(), offer->getUpdatee()));
+        deliverEventQueue.start(); 
+    }
+    // Process each frame through the error checker.
+    if (error.isUnresolved()) {
+        error.delivered(e);
+        while (error.canProcess())  // There is a frame ready to process.
+            processFrame(error.getNext(), l);
+    }
+    else 
+        processFrame(e, l);
 }
 
-Cluster::MemberList Cluster::getMembers() const {
-    Mutex::ScopedLock l(lock);
-    MemberList result(members.size());
-    std::transform(members.begin(), members.end(), result.begin(),
-                   boost::bind(&MemberMap::value_type::second, _1));
-    return result;        
+
+void Cluster::processFrame(const EventFrame& e, Lock& l) {
+    if (e.isCluster()) {
+        QPID_LOG(trace, *this << " DLVR: " << e);
+        ClusterDispatcher dispatch(*this, e.connectionId.getMember(), l);
+        if (!framing::invoke(dispatch, *e.frame.getBody()).wasHandled())
+            throw Exception(QPID_MSG("Invalid cluster control"));
+    }
+    else if (state >= CATCHUP) {
+        map.incrementFrameSeq();
+        ConnectionPtr connection = getConnection(e, l);
+        if (connection) {
+            QPID_LOG(trace, *this << " DLVR " << map.getFrameSeq() << ":  " << e);
+            connection->deliveredFrame(e);
+        }
+        else
+            QPID_LOG(trace, *this << " DROP (no connection): " << e);
+    }
+    else // Drop connection frames while state < CATCHUP
+        QPID_LOG(trace, *this << " DROP (joining): " << e);
 }
 
-// ################ HERE - leaking shadow connections.
-// FIXME aconway 2008-08-11: revisit memory management for shadow
-// connections, what if the Connection is closed other than via
-// disconnect? Dangling pointer in shadow map. Use ptr_map for shadow
-// map, add deleted state to ConnectionInterceptor? Interceptors need
-// to know about map? Check how Connections can be deleted.
+// Called in deliverFrameQueue thread
+ConnectionPtr Cluster::getConnection(const EventFrame& e, Lock&) {
+    ConnectionId id = e.connectionId;
+    ConnectionMap::iterator i = connections.find(id);
+    if (i != connections.end()) return i->second;
+    ConnectionPtr cp;
+    // If the frame is an announcement for a new connection, add it.
+    if (e.frame.getBody() && e.frame.getMethod() &&
+        e.frame.getMethod()->isA<ClusterConnectionAnnounceBody>())
+    {
+        if (id.getMember() == self)  { // Announces one of my own
+            cp = localConnections.getErase(id);
+            assert(cp); 
+        }
+        else {              // New remote connection, create a shadow.
+            std::ostringstream mgmtId;
+            unsigned int ssf;
+            const ClusterConnectionAnnounceBody *announce = castAnnounce(e.frame.getBody());
+
+            mgmtId << id;
+            ssf = (announce && announce->hasSsf()) ? announce->getSsf() : 0;
+            QPID_LOG(debug, *this << "new connection's ssf =" << ssf );
+            cp = new Connection(*this, shadowOut, mgmtId.str(), id, ssf );
+        }
+        connections.insert(ConnectionMap::value_type(id, cp));
+    }
+    return cp;
+}
 
-ConnectionInterceptor* Cluster::getShadowConnection(const Cpg::Id& member, void* remotePtr) {
-    ShadowConnectionId id(member, remotePtr);
-    ShadowConnectionMap::iterator i = shadowConnectionMap.find(id);
-    if (i == shadowConnectionMap.end()) { // A new shadow connection.
-        std::ostringstream os;
-        os << name << ":"  << member << ":" << remotePtr;
-        assert(broker);
-        broker::Connection* c = new broker::Connection(&shadowOut, *broker, os.str());
-        ShadowConnectionMap::value_type value(id, new ConnectionInterceptor(*c, *this, id));
-        i = shadowConnectionMap.insert(value).first;
+Cluster::ConnectionVector Cluster::getConnections(Lock&) {
+    ConnectionVector result(connections.size());
+    std::transform(connections.begin(), connections.end(), result.begin(),
+                   boost::bind(&ConnectionMap::value_type::second, _1));
+    return result;
+}
+  
+struct AddrList {
+    const cpg_address* addrs;
+    int count;
+    const char *prefix, *suffix;
+    AddrList(const cpg_address* a, int n, const char* p="", const char* s="")
+        : addrs(a), count(n), prefix(p), suffix(s) {}
+};
+
+ostream& operator<<(ostream& o, const AddrList& a) {
+    if (!a.count) return o;
+    o << a.prefix;
+    for (const cpg_address* p = a.addrs; p < a.addrs+a.count; ++p) {
+        const char* reasonString;
+        switch (p->reason) {
+          case CPG_REASON_JOIN: reasonString =  "(joined) "; break;
+          case CPG_REASON_LEAVE: reasonString =  "(left) "; break;
+          case CPG_REASON_NODEDOWN: reasonString =  "(node-down) "; break;
+          case CPG_REASON_NODEUP: reasonString =  "(node-up) "; break;
+          case CPG_REASON_PROCDOWN: reasonString =  "(process-down) "; break;
+          default: reasonString = " ";
+        }
+        qpid::cluster::MemberId member(*p);
+        o << member << reasonString;
     }
-    return i->second;
+    return o << a.suffix;
 }
 
-void Cluster::deliver(
+void Cluster::configChange ( 
     cpg_handle_t /*handle*/,
-    cpg_name* /*group*/,
-    uint32_t nodeid,
-    uint32_t pid,
-    void* msg,
-    int msg_len)
+    const cpg_name */*group*/,
+    const cpg_address *current, int nCurrent,
+    const cpg_address *left, int nLeft,
+    const cpg_address *joined, int nJoined)
 {
-    Id from(nodeid, pid);
-    try {
-        Buffer buf(static_cast<char*>(msg), msg_len);
-        while (buf.available() > 0) {
-            AMQFrame frame;
-            if (!frame.decode(buf))  // Not enough data.
-                throw Exception("Received incomplete cluster event.");
-            void* connection;
-            decodePtr(buf, connection);
-            deliverQueue.push(Message(frame, from, connection));
+    Mutex::ScopedLock l(lock);
+    QPID_LOG(notice, *this << " membership change: "
+             << AddrList(current, nCurrent) << "("
+             << AddrList(joined, nJoined, "joined: ")
+             << AddrList(left, nLeft, "left: ")
+             << ")");
+    std::string addresses;
+    for (const cpg_address* p = current; p < current+nCurrent; ++p) 
+        addresses.append(MemberId(*p).str());
+    deliverEvent(Event::control(ClusterConfigChangeBody(ProtocolVersion(), addresses), self));
+}
+
+void Cluster::setReady(Lock&) {
+    state = READY;
+    if (mgmtObject!=0) mgmtObject->set_status("ACTIVE");
+    mcast.setReady();
+    broker.getQueueEvents().enable();
+}
+
+void Cluster::initMapCompleted(Lock& l) {
+    // Called on completion of the initial status map. 
+    QPID_LOG(debug, *this << " initial status map complete. ");
+    if (state == INIT) {
+        // We have status for all members so we can make join descisions.
+        initMap.checkConsistent();
+        elders = initMap.getElders();
+        QPID_LOG(debug, *this << " elders: " << elders);
+        if (!elders.empty()) { // I'm not the elder, I don't handle links & replication.
+            broker.getLinks().setPassive(true);
+            broker.getQueueEvents().disable();
+            QPID_LOG(info, *this << " not active for links.");
+        }
+        else {
+            QPID_LOG(info, this << " active for links.");
+        }
+        setClusterId(initMap.getClusterId(), l);
+        if (store.hasStore()) store.dirty(clusterId);
+
+        if (initMap.isUpdateNeeded())  { // Joining established cluster.
+            broker.setRecovery(false); // Ditch my current store.
+            broker.setClusterUpdatee(true);
+            state = JOINER;
+        }
+        else {                  // I can go ready.
+            discarding = false;
+            setReady(l);
+            memberUpdate(l);
         }
+        QPID_LOG(debug, *this << "Initialization complete");
+    }
+}
+
+void Cluster::configChange(const MemberId&, const std::string& configStr, Lock& l) {
+    if (state == LEFT) return;
+
+    MemberSet config = decodeMemberSet(configStr);
+    elders = intersection(elders, config);
+    if (elders.empty() && INIT < state && state < CATCHUP) {
+        QPID_LOG(critical, "Cannot update, all potential updaters left the cluster.");
+        leave(l);
+        return;
     }
+    bool memberChange = map.configChange(config);
+
+    // Update initital status for new members joining.
+    initMap.configChange(config);
+    if (initMap.isResendNeeded()) {
+        mcast.mcastControl(
+            ClusterInitialStatusBody(
+                ProtocolVersion(), CLUSTER_VERSION, state > INIT, clusterId, 
+                store.getState(), store.getShutdownId()
+            ),
+            self);
+    }
+    if (initMap.transitionToComplete()) initMapCompleted(l);
+
+    if (state >= CATCHUP && memberChange) {
+        memberUpdate(l);
+        if (elders.empty()) {
+            // We are the oldest, reactive links if necessary
+            QPID_LOG(info, this << " becoming active for links.");
+            broker.getLinks().setPassive(false);
+        }
+    }
+}
+
+void Cluster::makeOffer(const MemberId& id, Lock& ) {
+    if (state == READY && map.isJoiner(id)) {
+        state = OFFER;
+        QPID_LOG(info, *this << " send update-offer to " << id);
+        mcast.mcastControl(ClusterUpdateOfferBody(ProtocolVersion(), id), self);
+    }
+}
+
+// Called from Broker::~Broker when broker is shut down.  At this
+// point we know the poller has stopped so no poller callbacks will be
+// invoked. We must ensure that CPG has also shut down so no CPG
+// callbacks will be invoked.
+// 
+void Cluster::brokerShutdown()  {
+    try { cpg.shutdown(); }
     catch (const std::exception& e) {
-        // FIXME aconway 2008-01-30: exception handling.
-        QPID_LOG(critical, "Error in cluster deliver: " << e.what());
-        assert(0);
-        throw;
+        QPID_LOG(error, *this << " shutting down CPG: " << e.what());
     }
+    delete this;
+}
+
+void Cluster::updateRequest(const MemberId& id, const std::string& url, Lock& l) {
+    map.updateRequest(id, url);
+    makeOffer(id, l);
 }
 
-void Cluster::deliverQueueCb(const MessageQueue::iterator& begin,
-                             const MessageQueue::iterator& end)
+void Cluster::initialStatus(const MemberId& member, uint32_t version, bool active,
+                            const framing::Uuid& id, 
+                            framing::cluster::StoreState store,
+                            const framing::Uuid& shutdownId,
+                            Lock& l)
 {
-    for (MessageQueue::iterator i = begin; i != end; ++i) {
-        AMQFrame& frame(i->frame);
-        Id from(i->from);
-        ConnectionInterceptor* connection = reinterpret_cast<ConnectionInterceptor*>(i->connection);
-        try {
-            QPID_LOG(trace, "DLVR [" << from << " " << connection << "] " << frame);
-
-            if (!broker) {
-                QPID_LOG(warning, "Unexpected DLVR, already left the cluster.");
-                return;
-            }
-            if (connection && from != self) // Look up shadow for remote connections
-                connection = getShadowConnection(from, connection);
+    if (version != CLUSTER_VERSION) {
+        QPID_LOG(critical, *this << " incompatible cluster versions " <<
+                 version << " != " << CLUSTER_VERSION);
+        leave(l);
+        return;
+    }
+    initMap.received(
+        member,
+        ClusterInitialStatusBody(ProtocolVersion(), version, active, id, store, shutdownId)
+    );
+    if (initMap.transitionToComplete()) initMapCompleted(l);
+}
 
-            if (frame.getMethod() && frame.getMethod()->amqpClassId() == CLUSTER_CLASS_ID) 
-                handleMethod(from, connection, *frame.getMethod());
-            else 
-                connection->deliver(frame);
+void Cluster::ready(const MemberId& id, const std::string& url, Lock& l) {
+    if (map.ready(id, Url(url))) 
+        memberUpdate(l);
+    if (state == CATCHUP && id == self) {
+        setReady(l);
+        QPID_LOG(notice, *this << " caught up.");
+    }
+}
+
+void Cluster::updateOffer(const MemberId& updater, uint64_t updateeInt, Lock& l) {
+    // NOTE: deliverEventQueue has been stopped at the update offer by
+    // deliveredEvent in case an update is required.
+    if (state == LEFT) return;
+    MemberId updatee(updateeInt);
+    boost::optional<Url> url = map.updateOffer(updater, updatee);
+    if (updater == self) {
+        assert(state == OFFER);
+        if (url)               // My offer was first.
+            updateStart(updatee, *url, l);
+        else {                  // Another offer was first.
+            QPID_LOG(info, *this << " cancelled offer to " << updatee << " unstall");
+            setReady(l);
+            makeOffer(map.firstJoiner(), l); // Maybe make another offer.
+            deliverEventQueue.start(); // Go back to normal processing
         }
-        catch (const std::exception& e) {
-            // FIXME aconway 2008-01-30: exception handling.
-            QPID_LOG(critical, "Error in cluster deliverQueueCb: " << e.what());
-            assert(0);
-            throw;
+    }
+    else if (updatee == self && url) {
+        assert(state == JOINER);
+        state = UPDATEE;
+        QPID_LOG(notice, *this << " receiving update from " << updater);
+        checkUpdateIn(l);
+    }
+    else {
+        QPID_LOG(debug,*this << " unstall, ignore update " << updater
+                 << " to " << updatee);
+        deliverEventQueue.start(); // Not involved in update.
+    }
+}
+
+static client::ConnectionSettings connectionSettings(const ClusterSettings& settings) {
+    client::ConnectionSettings cs;
+    cs.username = settings.username;
+    cs.password = settings.password;
+    cs.mechanism = settings.mechanism;
+    return cs;
+}
+
+void Cluster::retractOffer(const MemberId& updater, uint64_t updateeInt, Lock& l) {
+    // An offer was received while handling an error, and converted to a retract.
+    // Behavior is very similar to updateOffer.
+    if (state == LEFT) return;
+    MemberId updatee(updateeInt);
+    boost::optional<Url> url = map.updateOffer(updater, updatee);
+    if (updater == self) {
+        assert(state == OFFER);
+        if (url)  {             // My offer was first.
+            if (updateThread.id())
+                updateThread.join(); // Join the previous updateThread to avoid leaks.
+            updateThread = Thread(new RetractClient(*url, connectionSettings(settings)));
         }
+        setReady(l);
+        makeOffer(map.firstJoiner(), l); // Maybe make another offer.
+        // Don't unstall the event queue, that was already done in deliveredFrame
     }
+    QPID_LOG(debug,*this << " retracted offer " << updater << " to " << updatee);
+}
+
+void Cluster::updateStart(const MemberId& updatee, const Url& url, Lock& l) {
+    // NOTE: deliverEventQueue is already stopped at the stall point by deliveredEvent.
+    if (state == LEFT) return;
+    assert(state == OFFER);
+    state = UPDATER;
+    QPID_LOG(notice, *this << " sending update to " << updatee << " at " << url);
+    if (updateThread.id())
+        updateThread.join(); // Join the previous updateThread to avoid leaks.
+    updateThread = Thread(
+        new UpdateClient(self, updatee, url, broker, map, *expiryPolicy,
+                         getConnections(l), decoder,
+                         boost::bind(&Cluster::updateOutDone, this),
+                         boost::bind(&Cluster::updateOutError, this, _1),
+                         connectionSettings(settings)));
 }
 
-// Handle cluster methods
-// FIXME aconway 2008-07-11: Generate/template a better dispatch mechanism.
-void Cluster::handleMethod(Id from, ConnectionInterceptor* connection, AMQMethodBody& method) {
-    assert(method.amqpClassId() == CLUSTER_CLASS_ID);
-    switch (method.amqpMethodId()) {
-      case CLUSTER_NOTIFY_METHOD_ID: {
-          ClusterNotifyBody& notify=static_cast<ClusterNotifyBody&>(method);
-          Mutex::ScopedLock l(lock);
-          members[from].url=notify.getUrl();
-          lock.notifyAll();
-          break;
-      }
-      case CLUSTER_CONNECTION_CLOSE_METHOD_ID: {
-          if (!connection->isLocal())
-              shadowConnectionMap.erase(connection->getShadowId());
-          else
-              localConnectionSet.erase(connection);
-          connection->deliverClosed();
-          break;
-      }
-      case CLUSTER_CONNECTION_DO_OUTPUT_METHOD_ID: {
-          connection->deliverDoOutput();
-          break;
-      }
+// Called in update thread.
+void Cluster::updateInDone(const ClusterMap& m) {
+    Lock l(lock);
+    updatedMap = m;
+    checkUpdateIn(l);
+}
+
+void Cluster::updateInRetracted() {
+    Lock l(lock);
+    updateRetracted = true;
+    map.clearStatus();
+    checkUpdateIn(l);
+}
+
+void Cluster::checkUpdateIn(Lock& l) {
+    if (state != UPDATEE) return; // Wait till we reach the stall point.
+    if (updatedMap) { // We're up to date
+        map = *updatedMap;
+        memberUpdate(l);
+        mcast.mcastControl(ClusterReadyBody(ProtocolVersion(), myUrl.str()), self);
+        state = CATCHUP;
+        broker.setClusterUpdatee(false);
+        discarding = false;     // ok to set, we're stalled for update.
+        QPID_LOG(notice, *this << " update complete, starting catch-up.");
+        deliverEventQueue.start();
+    }
+    else if (updateRetracted) { // Update was retracted, request another update
+        updateRetracted = false;
+        state = JOINER;
+        QPID_LOG(notice, *this << " update retracted, sending new update request.");
+        mcast.mcastControl(ClusterUpdateRequestBody(ProtocolVersion(), myUrl.str()), self);
+        deliverEventQueue.start();
+    }
+}
+
+void Cluster::updateOutDone() {
+    Monitor::ScopedLock l(lock);
+    updateOutDone(l);
+}
+
+void Cluster::updateOutDone(Lock& l) {
+    QPID_LOG(notice, *this << " update sent");
+    assert(state == UPDATER);
+    state = READY;
+     deliverEventQueue.start();       // Start processing events again.
+    makeOffer(map.firstJoiner(), l); // Try another offer
+}
+
+void Cluster::updateOutError(const std::exception& e)  {
+    Monitor::ScopedLock l(lock);
+    QPID_LOG(error, *this << " error sending update: " << e.what());    
+    updateOutDone(l);
+}
+
+void Cluster ::shutdown(const MemberId& , const Uuid& id, Lock& l) {
+    QPID_LOG(notice, *this << " cluster shut down by administrator.");
+    if (store.hasStore()) store.clean(Uuid(id));
+    leave(l);
+}
+
+ManagementObject* Cluster::GetManagementObject() const { return mgmtObject; }
+
+Manageable::status_t Cluster::ManagementMethod (uint32_t methodId, Args& args, string&) {
+    Lock l(lock);
+    QPID_LOG(debug, *this << " managementMethod [id=" << methodId << "]");
+    switch (methodId) {
+      case _qmf::Cluster::METHOD_STOPCLUSTERNODE :
+        {
+            _qmf::ArgsClusterStopClusterNode& iargs = (_qmf::ArgsClusterStopClusterNode&) args;
+            stringstream stream;
+            stream << self;
+            if (iargs.i_brokerId == stream.str())
+                stopClusterNode(l);
+        }
+        break;
+      case _qmf::Cluster::METHOD_STOPFULLCLUSTER :
+        stopFullCluster(l);
+        break;
       default:
-        assert(0);
+        return Manageable::STATUS_UNKNOWN_METHOD;
     }
+    return Manageable::STATUS_OK;
 }
 
-void Cluster::configChange(
-    cpg_handle_t /*handle*/,
-    cpg_name */*group*/,
-    cpg_address *current, int nCurrent,
-    cpg_address *left, int nLeft,
-    cpg_address */*joined*/, int nJoined)
-{
-    Mutex::ScopedLock l(lock);
-    for (int i = 0; i < nLeft; ++i)  
-        members.erase(left[i]);
-    for(int j = 0; j < nCurrent; ++j) 
-        members[current[j]].id = current[j];
-    QPID_LOG(debug, "Cluster members: " << nCurrent << " ("<< nLeft << " left, " << nJoined << " joined):"
-             << members);
-    assert(members.size() == size_t(nCurrent));
-    if (members.find(self) == members.end()) 
-        broker = 0;       // We have left the group, this is the final config change.
-    lock.notifyAll();     // Threads waiting for membership changes.  
+void Cluster::stopClusterNode(Lock& l) {
+    QPID_LOG(notice, *this << " cluster member stopped by administrator.");
+    leave(l);
 }
 
-void Cluster::dispatch(sys::DispatchHandle& h) {
-    cpg.dispatchAll();
-    h.rewatch();
+void Cluster::stopFullCluster(Lock& ) {
+    QPID_LOG(notice, *this << " shutting down cluster " << name);
+    mcast.mcastControl(ClusterShutdownBody(ProtocolVersion(), Uuid(true)), self);
 }
 
-void Cluster::disconnect(sys::DispatchHandle& h) {
-    h.stopWatch();
-    // FIXME aconway 2008-08-11: error handling if we are disconnected. 
-    // Kill the broker?
-    assert(0);
+void Cluster::memberUpdate(Lock& l) {
+    QPID_LOG(info, *this << " member update: " << map);
+    std::vector<Url> urls = getUrls(l);
+    std::vector<string> ids = getIds(l);
+    size_t size = urls.size();
+    failoverExchange->setUrls(urls);
+
+    if (size == 1 && lastSize > 1 && state >= CATCHUP) { 
+        QPID_LOG(notice, *this << " last broker standing, update queue policies");
+        lastBroker = true;
+        broker.getQueues().updateQueueClusterState(true);
+    }
+    else if (size > 1 && lastBroker) {
+        QPID_LOG(notice, *this << " last broker standing joined by " << size-1 << " replicas, updating queue policies" << size);
+        lastBroker = false;
+        broker.getQueues().updateQueueClusterState(false);
+    }
+    lastSize = size;
+
+    if (mgmtObject) {
+        mgmtObject->set_clusterSize(size); 
+        string urlstr;
+        for(std::vector<Url>::iterator iter = urls.begin(); iter != urls.end(); iter++ ) {
+            if (iter != urls.begin()) urlstr += ";";
+            urlstr += iter->str();
+        }
+        string idstr;
+        for(std::vector<string>::iterator iter = ids.begin(); iter != ids.end(); iter++ ) {
+            if (iter != ids.begin()) idstr += ";";
+            idstr += (*iter);
+        }
+        mgmtObject->set_members(urlstr);
+        mgmtObject->set_memberIDs(idstr);
+    }
+
+    // Close connections belonging to members that have left the cluster.
+    ConnectionMap::iterator i = connections.begin();
+    while (i != connections.end()) {
+        ConnectionMap::iterator j = i++;
+        MemberId m = j->second->getId().getMember();
+        if (m != self && !map.isMember(m)) {
+            j->second->getBrokerConnection().closed();
+            erase(j->second->getId(), l);
+        }
+    }
 }
 
-}} // namespace qpid::cluster
+std::ostream& operator<<(std::ostream& o, const Cluster& cluster) {
+    static const char* STATE[] = {
+        "INIT", "JOINER", "UPDATEE", "CATCHUP", "READY", "OFFER", "UPDATER", "LEFT"
+    };
+    assert(sizeof(STATE)/sizeof(*STATE) == Cluster::LEFT+1);
+    o << "cluster(" << cluster.self << " " << STATE[cluster.state];
+    if (cluster.error.isUnresolved()) o << "/error";
+    return o << ")";;
+}
 
+MemberId Cluster::getId() const {
+    return self;            // Immutable, no need to lock.
+}
 
+broker::Broker& Cluster::getBroker() const {
+    return broker; // Immutable,  no need to lock.
+}
+
+void Cluster::setClusterId(const Uuid& uuid, Lock&) {
+    clusterId = uuid;
+    if (mgmtObject) {
+        stringstream stream;
+        stream << self;
+        mgmtObject->set_clusterID(clusterId.str());
+        mgmtObject->set_memberID(stream.str());
+    }
+    QPID_LOG(debug, *this << " cluster-uuid = " << clusterId);
+}
+
+void Cluster::messageExpired(const MemberId&, uint64_t id, Lock&) {
+    expiryPolicy->deliverExpire(id);
+}
+
+void Cluster::errorCheck(const MemberId& from, uint8_t type, framing::SequenceNumber frameSeq, Lock&) {
+    // If we see an errorCheck here (rather than in the ErrorCheck
+    // class) then we have processed succesfully past the point of the
+    // error.
+    if (state >= CATCHUP) // Don't respond pre catchup, we don't know what happened
+        error.respondNone(from, type, frameSeq);
+}
 
+
+}} // namespace qpid::cluster