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:20:51 UTC
svn commit: r903866 - in /qpid/trunk/qpid/cpp/src: ./ qpid/broker/
qpid/management/ qpid/sys/
Author: aconway
Date: Wed Jan 27 22:20:51 2010
New Revision: 903866
URL: http://svn.apache.org/viewvc?rev=903866&view=rev
Log:
Added PeriodicTimer interface for periodic tasks that need cluster synchronization.
The ManagementAgent's periodic prociessing uses PeriodicTimer.
PeriodicTimerImpl is the default implementation for stand-alone
brokers, simple wrapper for sys::Timer.
Added:
qpid/trunk/qpid/cpp/src/qpid/sys/PeriodicTimer.h
qpid/trunk/qpid/cpp/src/qpid/sys/PeriodicTimerImpl.cpp
qpid/trunk/qpid/cpp/src/qpid/sys/PeriodicTimerImpl.h
Modified:
qpid/trunk/qpid/cpp/src/Makefile.am
qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h
qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.cpp
qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.h
qpid/trunk/qpid/cpp/src/qpid/sys/Timer.h
Modified: qpid/trunk/qpid/cpp/src/Makefile.am
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/Makefile.am?rev=903866&r1=903865&r2=903866&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/Makefile.am (original)
+++ qpid/trunk/qpid/cpp/src/Makefile.am Wed Jan 27 22:20:51 2010
@@ -471,6 +471,9 @@
qpid/sys/TimeoutHandler.h \
qpid/sys/Timer.cpp \
qpid/sys/Timer.h \
+ qpid/sys/PeriodicTimer.h \
+ qpid/sys/PeriodicTimerImpl.h \
+ qpid/sys/PeriodicTimerImpl.cpp \
qpid/sys/Waitable.h \
qpid/sys/alloca.h \
qpid/sys/uuid.h
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp?rev=903866&r1=903865&r2=903866&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp Wed Jan 27 22:20:51 2010
@@ -49,6 +49,7 @@
#include "qpid/sys/ConnectionInputHandlerFactory.h"
#include "qpid/sys/TimeoutHandler.h"
#include "qpid/sys/SystemInfo.h"
+#include "qpid/sys/PeriodicTimerImpl.h"
#include "qpid/Address.h"
#include "qpid/Url.h"
#include "qpid/Version.h"
@@ -136,6 +137,7 @@
Broker::Broker(const Broker::Options& conf) :
poller(new Poller),
+ periodicTimer(new sys::PeriodicTimerImpl(timer)),
config(conf),
managementAgent(conf.enableMgmt ? new ManagementAgent() : 0),
store(new NullMessageStore),
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h?rev=903866&r1=903865&r2=903866&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h Wed Jan 27 22:20:51 2010
@@ -49,6 +49,7 @@
#include "qpid/framing/ProtocolInitiation.h"
#include "qpid/sys/Runnable.h"
#include "qpid/sys/Timer.h"
+#include "qpid/sys/PeriodicTimer.h"
#include "qpid/RefCounted.h"
#include "qpid/broker/AclModule.h"
#include "qpid/sys/Mutex.h"
@@ -146,6 +147,7 @@
boost::shared_ptr<sys::Poller> poller;
sys::Timer timer;
+ std::auto_ptr<sys::PeriodicTimer> periodicTimer;
Options config;
std::auto_ptr<management::ManagementAgent> managementAgent;
ProtocolFactoryMap protocolFactories;
@@ -253,6 +255,8 @@
void setConnectionFactory(boost::shared_ptr<sys::ConnectionCodec::Factory> f) { factory = f; }
sys::Timer& getTimer() { return timer; }
+ sys::PeriodicTimer& getPeriodicTimer() { return *periodicTimer; }
+ void setPeriodicTimer(std::auto_ptr<sys::PeriodicTimer> pt) { periodicTimer = pt; }
boost::function<std::vector<Url> ()> getKnownBrokers;
Modified: qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.cpp?rev=903866&r1=903865&r2=903866&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.cpp Wed Jan 27 22:20:51 2010
@@ -91,10 +91,12 @@
dataDir = _dataDir;
interval = _interval;
broker = _broker;
- timer = &_broker->getTimer();
+ timer = &_broker->getPeriodicTimer();
threadPoolSize = _threads;
ManagementObject::maxThreads = threadPoolSize;
- timer->add (new Periodic(*this, interval));
+ timer->add (boost::bind(&ManagementAgent::periodicProcessing, this),
+ interval * sys::TIME_SEC,
+ "ManagementAgent::periodicProcessing");
// Get from file or generate and save to file.
if (dataDir.empty())
@@ -231,17 +233,6 @@
"console.event.1.0." + event.getPackageName() + "." + event.getEventName());
}
-ManagementAgent::Periodic::Periodic (ManagementAgent& _agent, uint32_t _seconds)
- : TimerTask (qpid::sys::Duration ((_seconds ? _seconds : 1) * qpid::sys::TIME_SEC)), agent(_agent) {}
-
-ManagementAgent::Periodic::~Periodic () {}
-
-void ManagementAgent::Periodic::fire ()
-{
- agent.timer->add (new Periodic (agent, agent.interval));
- agent.periodicProcessing ();
-}
-
void ManagementAgent::clientAdded (const std::string& routingKey)
{
if (routingKey.find("console") != 0)
@@ -332,6 +323,7 @@
{
#define BUFSIZE 65536
#define HEADROOM 4096
+ QPID_LOG(trace, "Management agent periodic processing")
Mutex::ScopedLock lock (userLock);
char msgChars[BUFSIZE];
uint32_t contentSize;
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=903866&r1=903865&r2=903866&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.h Wed Jan 27 22:20:51 2010
@@ -26,7 +26,7 @@
#include "qpid/broker/Exchange.h"
#include "qpid/framing/Uuid.h"
#include "qpid/sys/Mutex.h"
-#include "qpid/sys/Timer.h"
+#include "qpid/sys/PeriodicTimer.h"
#include "qpid/broker/ConnectionToken.h"
#include "qpid/management/ManagementObject.h"
#include "qpid/management/ManagementEvent.h"
@@ -105,15 +105,6 @@
void importSchemas(framing::Buffer& inBuf);
private:
- struct Periodic : public qpid::sys::TimerTask
- {
- ManagementAgent& agent;
-
- Periodic (ManagementAgent& agent, uint32_t seconds);
- virtual ~Periodic ();
- void fire ();
- };
-
// Storage for tracking remote management agents, attached via the client
// management agent API.
//
@@ -205,7 +196,7 @@
std::string dataDir;
uint16_t interval;
qpid::broker::Broker* broker;
- qpid::sys::Timer* timer;
+ qpid::sys::PeriodicTimer* timer;
uint16_t bootSequence;
uint32_t nextObjectId;
uint32_t brokerBank;
Added: qpid/trunk/qpid/cpp/src/qpid/sys/PeriodicTimer.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/PeriodicTimer.h?rev=903866&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/PeriodicTimer.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/PeriodicTimer.h Wed Jan 27 22:20:51 2010
@@ -0,0 +1,56 @@
+#ifndef QPID_SYS_PERIODICTIMER_H
+#define QPID_SYS_PERIODICTIMER_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 "Timer.h"
+#include <boost/function.hpp>
+
+namespace qpid {
+namespace sys {
+
+/**
+ * Interface of a timer for periodic tasks that should be synchronized
+ * across all brokers in a periodic. The standalone broker
+ * implementation simply wraps qpid::sys::Timer. The clustered broker
+ * implementation synchronizes execution of periodic tasks on all
+ * periodic members.
+ */
+class PeriodicTimer
+{
+ public:
+ typedef boost::function<void()> Task;
+
+ QPID_COMMON_EXTERN virtual ~PeriodicTimer() {}
+
+ /**
+ * Add a named task to be executed at the given period.
+ *
+ * The task registered under the same name will be executed on
+ * all brokers at the given period.
+ */
+ virtual void add(const Task& task, Duration period, const std::string& taskName) = 0;
+
+};
+}} // namespace qpid::sys
+
+#endif /*!QPID_SYS_PERIODICTIMER_H*/
Added: qpid/trunk/qpid/cpp/src/qpid/sys/PeriodicTimerImpl.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/PeriodicTimerImpl.cpp?rev=903866&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/PeriodicTimerImpl.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/PeriodicTimerImpl.cpp Wed Jan 27 22:20:51 2010
@@ -0,0 +1,49 @@
+/*
+ *
+ * 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"
+
+namespace qpid {
+namespace sys {
+
+PeriodicTimerImpl::PeriodicTimerImpl(Timer& t) : timer(t) {}
+
+struct PeriodicTimerImpl::TaskImpl : public TimerTask {
+ Timer& timer;
+ Task task;
+
+ TaskImpl(Timer& timer_, const Task& task_, Duration period) :
+ TimerTask(period), timer(timer_), task(task_) {}
+
+ void fire() {
+ task();
+ setupNextFire();
+ timer.add(this);
+ }
+};
+
+void PeriodicTimerImpl::add(
+ const Task& task, Duration period, const std::string& /*taskName*/
+)
+{
+ timer.add(new TaskImpl(timer, task, period));
+}
+
+}} // namespace qpid::sys
Added: qpid/trunk/qpid/cpp/src/qpid/sys/PeriodicTimerImpl.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/PeriodicTimerImpl.h?rev=903866&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/PeriodicTimerImpl.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/PeriodicTimerImpl.h Wed Jan 27 22:20:51 2010
@@ -0,0 +1,45 @@
+#ifndef QPID_SYS_PERIODICTIMERIMPL_H
+#define QPID_SYS_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 "PeriodicTimer.h"
+
+namespace qpid {
+namespace sys {
+
+/**
+ * Standalone broker implementation of PeriodicTimer.
+ */
+class PeriodicTimerImpl : public PeriodicTimer
+{
+ public:
+ PeriodicTimerImpl(Timer& timer);
+ void add(const Task& task, Duration period, const std::string& taskName);
+
+ private:
+ struct TaskImpl;
+ Timer& timer;
+};
+}} // namespace qpid::sys
+
+#endif /*!QPID_SYS_PERIODICTIMER_H*/
Modified: qpid/trunk/qpid/cpp/src/qpid/sys/Timer.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/Timer.h?rev=903866&r1=903865&r2=903866&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/Timer.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/Timer.h Wed Jan 27 22:20:51 2010
@@ -69,6 +69,12 @@
bool operator<(const boost::intrusive_ptr<TimerTask>& a,
const boost::intrusive_ptr<TimerTask>& b);
+/**
+ A timer to trigger tasks that are local to one broker.
+
+ For periodic tasks that should be synchronized across all brokers
+ in a cluster, use qpid::sys::PeriodicTimer.
+ */
class Timer : private Runnable {
qpid::sys::Monitor monitor;
std::priority_queue<boost::intrusive_ptr<TimerTask> > tasks;
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org