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 2010/01/28 22:38:20 UTC
svn commit: r904270 - in /qpid/trunk/qpid: cpp/src/qpid/agent/
cpp/src/qpid/cluster/ cpp/src/qpid/management/ cpp/xml/ specs/
Author: aconway
Date: Thu Jan 28 21:38:19 2010
New Revision: 904270
URL: http://svn.apache.org/viewvc?rev=904270&view=rev
Log:
QPID-2357 Broker boot sequence doesn't synchronize when clustered - patch from John Dunning
Modified:
qpid/trunk/qpid/cpp/src/qpid/agent/ManagementAgentImpl.h
qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp
qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h
qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp
qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.h
qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.h
qpid/trunk/qpid/cpp/xml/cluster.xml
qpid/trunk/qpid/specs/management-schema.xml
Modified: qpid/trunk/qpid/cpp/src/qpid/agent/ManagementAgentImpl.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/agent/ManagementAgentImpl.h?rev=904270&r1=904269&r2=904270&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/agent/ManagementAgentImpl.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/agent/ManagementAgentImpl.h Thu Jan 28 21:38:19 2010
@@ -82,6 +82,13 @@
uint16_t getInterval() { return interval; }
void periodicProcessing();
+ // these next are here to support the hot-wiring of state between clustered brokers
+ uint64_t getNextObjectId(void) { return nextObjectId; }
+ void setNextObjectId(uint64_t o) { nextObjectId = o; }
+
+ uint16_t getBootSequence(void) { return bootSequence; }
+ void setBootSequence(uint16_t b) { bootSequence = b; }
+
private:
struct SchemaClassKey {
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp?rev=904270&r1=904269&r2=904270&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp Thu Jan 28 21:38:19 2010
@@ -513,5 +513,19 @@
QPID_LOG(debug, cluster << " updated management schemas");
}
+//
+// This is the handler for incoming managementsetup messages.
+//
+void Connection::managementSetupState(uint64_t objectNum, uint16_t bootSequence)
+{
+ QPID_LOG(debug, "Running managementsetup state handler, new objectnum "
+ << objectNum << " seq " << bootSequence);
+ management::ManagementAgent* agent = cluster.getBroker().getManagementAgent();
+ if (!agent)
+ throw Exception(QPID_MSG("Management schema update but no management agent."));
+ agent->setNextObjectId(objectNum);
+ agent->setBootSequence(bootSequence);
+}
+
}} // Namespace qpid::cluster
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h?rev=904270&r1=904269&r2=904270&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h Thu Jan 28 21:38:19 2010
@@ -164,6 +164,7 @@
void addQueueListener(const std::string& queue, uint32_t listener);
void managementSchema(const std::string& data);
+ void managementSetupState(uint64_t objectNum, uint16_t bootSequence);
uint32_t getSsf() const { return connectionCtor.ssf; }
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp?rev=904270&r1=904269&r2=904270&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp Thu Jan 28 21:38:19 2010
@@ -127,6 +127,17 @@
QPID_LOG(debug, updaterId << " updating state to " << updateeId
<< " at " << updateeUrl);
Broker& b = updaterBroker;
+
+ //
+ // Bash the state of the slave into conformance with ours. The
+ // goal here is to get his state arranged so as to mimic our
+ // state, w/r/t object ID creation. Currently, that means that we
+ // propagate our boot seq and object UID counter to him so that
+ // subsequently created objects on his side will track what's on
+ // our side.
+ //
+ updateManagementSetupState(b);
+
b.getExchanges().eachExchange(boost::bind(&UpdateClient::updateExchange, this, _1));
b.getQueues().eachQueue(boost::bind(&UpdateClient::updateNonExclusiveQueue, this, _1));
@@ -173,6 +184,22 @@
}
} // namespace
+//
+// Propagate the management setup state block, currently consisting of
+// object number counter and boot sequence counter, to the slave.
+//
+void UpdateClient::updateManagementSetupState(Broker & b)
+{
+ qmf::org::apache::qpid::broker::ManagementSetupState mss(b.getManagementAgent(), 0);
+ mss.set_objectNum(b.getManagementAgent()->getNextObjectId());
+ mss.set_bootSequence(b.getManagementAgent()->getBootSequence());
+
+ QPID_LOG(debug, updaterId << " updating management-setup-state " << mss.get_objectNum()
+ << " " << mss.get_bootSequence() << "\n");
+
+ ClusterConnectionProxy(session).managementSetupState(mss.get_objectNum(), mss.get_bootSequence());
+}
+
void UpdateClient::updateExchange(const boost::shared_ptr<Exchange>& ex) {
QPID_LOG(debug, updaterId << " updating exchange " << ex->getName());
ClusterConnectionProxy(session).exchange(encode(*ex));
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.h?rev=904270&r1=904269&r2=904270&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.h Thu Jan 28 21:38:19 2010
@@ -29,6 +29,7 @@
#include "qpid/client/AsyncSession.h"
#include "qpid/broker/SemanticState.h"
#include "qpid/sys/Runnable.h"
+#include "qmf/org/apache/qpid/broker/ManagementSetupState.h"
#include <boost/shared_ptr.hpp>
@@ -97,6 +98,7 @@
void updateConsumer(const broker::SemanticState::ConsumerImpl::shared_ptr&);
void updateQueueListeners(const boost::shared_ptr<broker::Queue>&);
void updateQueueListener(std::string& q, const boost::shared_ptr<broker::Consumer>& c);
+ void updateManagementSetupState(broker::Broker & b);
Numbering<broker::SemanticState::ConsumerImpl::shared_ptr> consumerNumbering;
MemberId updaterId;
Modified: qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.h?rev=904270&r1=904269&r2=904270&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.h Thu Jan 28 21:38:19 2010
@@ -104,6 +104,13 @@
/** Decode a serialized schemas and add to my schema cache */
void importSchemas(framing::Buffer& inBuf);
+ // these are in support of the managementSetup-state stuff, for synch'ing clustered brokers
+ uint64_t getNextObjectId(void) { return nextObjectId; }
+ void setNextObjectId(uint64_t o) { nextObjectId = o; }
+
+ uint16_t getBootSequence(void) { return bootSequence; }
+ void setBootSequence(uint16_t b) { bootSequence = b; }
+
private:
// Storage for tracking remote management agents, attached via the client
// management agent API.
Modified: qpid/trunk/qpid/cpp/xml/cluster.xml
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/xml/cluster.xml?rev=904270&r1=904269&r2=904270&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/xml/cluster.xml (original)
+++ qpid/trunk/qpid/cpp/xml/cluster.xml Thu Jan 28 21:38:19 2010
@@ -237,5 +237,13 @@
<control name="management-schema" code="0x35">
<field name="data" type="vbin32"/>
</control>
+
+ <!-- added by jrd. propagate a management-setup-state widget -->
+ <control name="management-setup-state" code="0x36">
+ <field name="objectNum" type="uint64"/>
+ <field name="bootSequence" type="uint16"/>
+ </control>
+
+
</class>
</amqp>
Modified: qpid/trunk/qpid/specs/management-schema.xml
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/specs/management-schema.xml?rev=904270&r1=904269&r2=904270&view=diff
==============================================================================
--- qpid/trunk/qpid/specs/management-schema.xml (original)
+++ qpid/trunk/qpid/specs/management-schema.xml Thu Jan 28 21:38:19 2010
@@ -326,6 +326,27 @@
<method name="close"/>
</class>
+ <!--
+ ===============================================================
+ ManagementSetupState
+ ===============================================================
+
+ This thing is used during cluster recovery operations (and maybe
+ eventually elsewhere) to transmit assorted state from one broker to
+ another. At present, the two data propagated are the object number
+ counter and boot sequence, both of which are used for creating
+ object ids for newly-created objects.
+
+ -->
+ <class name="ManagementSetupState">
+ <!-- for reasons that aren't clear (to me, anyhow) you have to say
+ access="RO" to get accessor methods defined. RC or RW don't do
+ it. Probably this is documented someplace, but I couldn't find
+ it. -jrd -->
+ <property name="objectNum" type="uint64" access="RO"/>
+ <property name="bootSequence" type="uint16" access="RO"/>
+ </class>
+
<eventArguments>
<arg name="altEx" type="sstr" desc="Name of the alternate exchange"/>
<arg name="args" type="map" desc="Supplemental arguments or parameters supplied"/>
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org