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/27 23:21:03 UTC
svn commit: r903867 - in /qpid/trunk/qpid/cpp: src/CMakeLists.txt
src/cluster.cmake src/cluster.mk src/qpid/cluster/Cluster.cpp
src/qpid/cluster/Cluster.h src/qpid/cluster/PeriodicTimerImpl.cpp
src/qpid/cluster/PeriodicTimerImpl.h xml/cluster.xml
Author: aconway
Date: Wed Jan 27 22:21:02 2010
New Revision: 903867
URL: http://svn.apache.org/viewvc?rev=903867&view=rev
Log:
Cluster implementation of PeriodicTimer.
The cluster implementation multicast periodic-timer controls and
executes the task when those controls are delivered, which is in the
cluster delivery thread context and so consistent across the cluster.
Added:
qpid/trunk/qpid/cpp/src/qpid/cluster/PeriodicTimerImpl.cpp
qpid/trunk/qpid/cpp/src/qpid/cluster/PeriodicTimerImpl.h
Modified:
qpid/trunk/qpid/cpp/src/CMakeLists.txt
qpid/trunk/qpid/cpp/src/cluster.cmake
qpid/trunk/qpid/cpp/src/cluster.mk
qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp
qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h
qpid/trunk/qpid/cpp/xml/cluster.xml
Modified: qpid/trunk/qpid/cpp/src/CMakeLists.txt
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/CMakeLists.txt?rev=903867&r1=903866&r2=903867&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/CMakeLists.txt (original)
+++ qpid/trunk/qpid/cpp/src/CMakeLists.txt Wed Jan 27 22:21:02 2010
@@ -573,6 +573,7 @@
qpid/sys/Runnable.cpp
qpid/sys/Shlib.cpp
qpid/sys/Timer.cpp
+ qpid/sys/PeriodicTimerImpl.cpp
)
add_library (qpidcommon SHARED ${qpidcommon_SOURCES})
Modified: qpid/trunk/qpid/cpp/src/cluster.cmake
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/cluster.cmake?rev=903867&r1=903866&r2=903867&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/cluster.cmake (original)
+++ qpid/trunk/qpid/cpp/src/cluster.cmake Wed Jan 27 22:21:02 2010
@@ -129,6 +129,8 @@
qpid/cluster/MemberSet.h
qpid/cluster/MemberSet.cpp
qpid/cluster/types.h
+ qpid/cluster/PeriodicTimerImpl.h
+ qpid/cluster/PeriodicTimerImpl.cpp
qpid/cluster/StoreStatus.h
qpid/cluster/StoreStatus.cpp
)
Modified: qpid/trunk/qpid/cpp/src/cluster.mk
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/cluster.mk?rev=903867&r1=903866&r2=903867&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/cluster.mk (original)
+++ qpid/trunk/qpid/cpp/src/cluster.mk Wed Jan 27 22:21:02 2010
@@ -88,6 +88,8 @@
qpid/cluster/MemberSet.h \
qpid/cluster/MemberSet.cpp \
qpid/cluster/types.h \
+ qpid/cluster/PeriodicTimerImpl.h \
+ qpid/cluster/PeriodicTimerImpl.cpp \
qpid/cluster/StoreStatus.h \
qpid/cluster/StoreStatus.cpp
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp?rev=903867&r1=903866&r2=903867&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp Wed Jan 27 22:21:02 2010
@@ -112,6 +112,7 @@
#include "qpid/cluster/RetractClient.h"
#include "qpid/cluster/FailoverExchange.h"
#include "qpid/cluster/UpdateExchange.h"
+#include "qpid/cluster/PeriodicTimerImpl.h"
#include "qpid/assert.h"
#include "qmf/org/apache/qpid/cluster/ArgsClusterStopClusterNode.h"
@@ -136,6 +137,7 @@
#include "qpid/framing/ClusterUpdateRequestBody.h"
#include "qpid/framing/ClusterConnectionAnnounceBody.h"
#include "qpid/framing/ClusterErrorCheckBody.h"
+#include "qpid/framing/ClusterPeriodicTimerBody.h"
#include "qpid/framing/MessageTransferBody.h"
#include "qpid/log/Helpers.h"
#include "qpid/log/Statement.h"
@@ -177,7 +179,7 @@
* Currently use SVN revision to avoid clashes with versions from
* different branches.
*/
-const uint32_t Cluster::CLUSTER_VERSION = 896973;
+const uint32_t Cluster::CLUSTER_VERSION = 903171;
struct ClusterDispatcher : public framing::AMQP_AllOperations::ClusterHandler {
qpid::cluster::Cluster& cluster;
@@ -207,6 +209,9 @@
void errorCheck(uint8_t type, const framing::SequenceNumber& frameSeq) {
cluster.errorCheck(member, type, frameSeq, l);
}
+ void periodicTimer(const std::string& name) {
+ cluster.periodicTimer(member, name, l);
+ }
void shutdown(const Uuid& id) { cluster.shutdown(member, id, l); }
@@ -245,6 +250,10 @@
updateRetracted(false),
error(*this)
{
+ // FIXME aconway 2010-01-26: must be done before management registers with timer.
+ broker.setPeriodicTimer(
+ std::auto_ptr<sys::PeriodicTimer>(new PeriodicTimerImpl(*this)));
+
mAgent = broker.getManagementAgent();
if (mAgent != 0){
_qmf::Package packageInit(mAgent);
@@ -952,5 +961,13 @@
error.respondNone(from, type, frameSeq);
}
+void Cluster::periodicTimer(const MemberId&, const std::string& , Lock&) {
+ // FIXME aconway 2010-01-26:
+}
+
+bool Cluster::isElder() const {
+ Mutex::ScopedLock l(lock);
+ return elders.empty();
+}
}} // namespace qpid::cluster
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h?rev=903867&r1=903866&r2=903867&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h Wed Jan 27 22:21:02 2010
@@ -117,6 +117,8 @@
UpdateReceiver& getUpdateReceiver() { return updateReceiver; }
+ bool isElder() const;
+
private:
typedef sys::Monitor::ScopedLock Lock;
@@ -161,6 +163,7 @@
void configChange(const MemberId&, const std::string& current, Lock& l);
void messageExpired(const MemberId&, uint64_t, Lock& l);
void errorCheck(const MemberId&, uint8_t type, SequenceNumber frameSeq, Lock&);
+ void periodicTimer(const MemberId&, const std::string& name, Lock&);
void shutdown(const MemberId&, const framing::Uuid& shutdownId, Lock&);
@@ -241,11 +244,6 @@
// Remaining members are protected by lock.
-
- // TODO aconway 2009-03-06: Most of these members are also only used in
- // deliverFrameQueue thread or during stall. Review and separate members
- // that require a lock, drop lock when not needed.
-
mutable sys::Monitor lock;
Added: qpid/trunk/qpid/cpp/src/qpid/cluster/PeriodicTimerImpl.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/PeriodicTimerImpl.cpp?rev=903867&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/PeriodicTimerImpl.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/PeriodicTimerImpl.cpp Wed Jan 27 22:21:02 2010
@@ -0,0 +1,78 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "PeriodicTimerImpl.h"
+#include "Cluster.h"
+#include "qpid/framing/ProtocolVersion.h"
+#include "qpid/framing/ClusterPeriodicTimerBody.h"
+
+namespace qpid {
+namespace cluster {
+
+PeriodicTimerImpl::PeriodicTimerImpl(Cluster& c) : cluster(c) {}
+
+PeriodicTimerImpl::TaskEntry::TaskEntry(
+ Cluster& c, const Task& t, sys::Duration d, const std::string& n)
+ : TimerTask(d), cluster(c), timer(c.getBroker().getTimer()),
+ task(t), name(n), inFlight(false)
+{}
+
+void PeriodicTimerImpl::TaskEntry::fire() {
+ sys::Mutex::ScopedLock l(lock);
+ // Only the elder mcasts.
+ // Don't mcast another if we haven't yet received the last one.
+ if (cluster.isElder() && !inFlight) {
+ inFlight = true;
+ cluster.getMulticast().mcastControl(
+ framing::ClusterPeriodicTimerBody(framing::ProtocolVersion(), name),
+ cluster.getId());
+ }
+ setupNextFire();
+ timer.add(this);
+}
+
+void PeriodicTimerImpl::TaskEntry::deliver() {
+ task();
+ sys::Mutex::ScopedLock l(lock);
+ inFlight = false;
+}
+
+
+void PeriodicTimerImpl::add(
+ const Task& task, sys::Duration period, const std::string& name)
+{
+ sys::Mutex::ScopedLock l(lock);
+ if (map.find(name) != map.end())
+ throw Exception(QPID_MSG("Cluster timer task name added twice: " << name));
+ map[name] = new TaskEntry(cluster, task, period, name);
+}
+
+void PeriodicTimerImpl::deliver(const std::string& name) {
+ Map::iterator i;
+ {
+ sys::Mutex::ScopedLock l(lock);
+ i = map.find(name);
+ if (i == map.end())
+ throw Exception(QPID_MSG("Cluster timer unknown task: " << name));
+ }
+ i->second->deliver();
+}
+
+}} // namespace qpid::cluster
Added: qpid/trunk/qpid/cpp/src/qpid/cluster/PeriodicTimerImpl.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/PeriodicTimerImpl.h?rev=903867&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/PeriodicTimerImpl.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/PeriodicTimerImpl.h Wed Jan 27 22:21:02 2010
@@ -0,0 +1,72 @@
+#ifndef QPID_CLUSTER_PERIODICTIMERIMPL_H
+#define QPID_CLUSTER_PERIODICTIMERIMPL_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/sys/PeriodicTimer.h"
+#include "qpid/sys/Mutex.h"
+#include <map>
+
+namespace qpid {
+namespace cluster {
+
+class Cluster;
+
+/**
+ * Cluster implementation of PeriodicTimer.
+ *
+ * All members run a periodic task, elder mcasts periodic-timer control.
+ * Actual task is executed on delivery of periodic-timer.
+ */
+class PeriodicTimerImpl : public sys::PeriodicTimer
+{
+ public:
+ PeriodicTimerImpl(Cluster& cluster);
+ void add(const Task& task, sys::Duration period, const std::string& taskName);
+ void deliver(const std::string& name);
+
+ private:
+
+ class TaskEntry : public sys::TimerTask {
+ public:
+ TaskEntry(Cluster&, const Task&, sys::Duration period, const std::string& name);
+ void fire();
+ void deliver();
+ private:
+ sys::Mutex lock;
+ Cluster& cluster;
+ sys::Timer& timer;
+ Task task;
+ std::string name;
+ bool inFlight;
+ };
+
+ typedef std::map<std::string, boost::intrusive_ptr<TaskEntry> > Map;
+ struct TaskImpl;
+
+ sys::Mutex lock;
+ Map map;
+ Cluster& cluster;
+};
+}} // namespace qpid::cluster
+
+#endif /*!QPID_CLUSTER_PERIODICTIMER_H*/
Modified: qpid/trunk/qpid/cpp/xml/cluster.xml
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/xml/cluster.xml?rev=903867&r1=903866&r2=903867&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/xml/cluster.xml (original)
+++ qpid/trunk/qpid/cpp/xml/cluster.xml Wed Jan 27 22:21:02 2010
@@ -95,6 +95,11 @@
<field name="frame-seq" type="sequence-no"/>
</control>
+ <!-- Synchronize periodic timer tasks across the cluster -->
+ <control name="periodic-timer" code="0x15">
+ <field name="name" type="str16"/>
+ </control>
+
<!-- Shut down the entire cluster -->
<control name="shutdown" code="0x20">
<field name="shutdown-id" type="uuid"/>
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org