You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kg...@apache.org on 2012/01/13 19:22:41 UTC
svn commit: r1231221 - in /qpid/trunk/qpid/cpp/src: qpid/broker/
qpid/cluster/ tests/
Author: kgiusti
Date: Fri Jan 13 18:22:40 2012
New Revision: 1231221
URL: http://svn.apache.org/viewvc?rev=1231221&view=rev
Log:
QPID-3630: remove deprecated rate limiting feature
Removed:
qpid/trunk/qpid/cpp/src/qpid/broker/RateFlowcontrol.h
qpid/trunk/qpid/cpp/src/tests/RateFlowcontrolTest.cpp
Modified:
qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h
qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionHandler.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionState.h
qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.h
qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp
qpid/trunk/qpid/cpp/src/tests/CMakeLists.txt
qpid/trunk/qpid/cpp/src/tests/Makefile.am
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=1231221&r1=1231220&r2=1231221&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp Fri Jan 13 18:22:40 2012
@@ -120,7 +120,6 @@ Broker::Options::Options(const std::stri
queueLimit(100*1048576/*100M default limit*/),
tcpNoDelay(false),
requireEncrypted(false),
- maxSessionRate(0),
asyncQueueEvents(false), // Must be false in a cluster.
qmf2Support(true),
qmf1Support(true),
@@ -160,7 +159,6 @@ Broker::Options::Options(const std::stri
("require-encryption", optValue(requireEncrypted), "Only accept connections that are encrypted")
("known-hosts-url", optValue(knownHosts, "URL or 'none'"), "URL to send as 'known-hosts' to clients ('none' implies empty list)")
("sasl-config", optValue(saslConfigPath, "DIR"), "gets sasl config info from nonstandard location")
- ("max-session-rate", optValue(maxSessionRate, "MESSAGES/S"), "Sets the maximum message rate per session (0=unlimited)")
("async-queue-events", optValue(asyncQueueEvents, "yes|no"), "Set Queue Events async, used for services like replication")
("default-flow-stop-threshold", optValue(queueFlowStopRatio, "PERCENT"), "Percent of queue's maximum capacity at which flow control is activated.")
("default-flow-resume-threshold", optValue(queueFlowResumeRatio, "PERCENT"), "Percent of queue's maximum capacity at which flow control is de-activated.")
@@ -339,12 +337,6 @@ Broker::Broker(const Broker::Options& co
knownBrokers.push_back(Url(conf.knownHosts));
}
- // check for and warn if deprecated features have been configured
- if (conf.maxSessionRate) {
- QPID_LOG(warning, "The 'max-session-rate' feature will be removed in a future release of QPID."
- " Queue-based flow control should be used instead.");
- }
-
} catch (const std::exception& /*e*/) {
finalize();
throw;
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=1231221&r1=1231220&r2=1231221&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h Fri Jan 13 18:22:40 2012
@@ -114,7 +114,6 @@ public:
bool requireEncrypted;
std::string knownHosts;
std::string saslConfigPath;
- uint32_t maxSessionRate;
bool asyncQueueEvents;
bool qmf2Support;
bool qmf1Support;
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionHandler.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionHandler.cpp?rev=1231221&r1=1231220&r2=1231221&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionHandler.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionHandler.cpp Fri Jan 13 18:22:40 2012
@@ -46,11 +46,9 @@ const std::string PLAIN = "PLAIN";
const std::string en_US = "en_US";
const std::string QPID_FED_LINK = "qpid.fed_link";
const std::string QPID_FED_TAG = "qpid.federation_tag";
-const std::string SESSION_FLOW_CONTROL("qpid.session_flow");
const std::string CLIENT_PROCESS_NAME("qpid.client_process");
const std::string CLIENT_PID("qpid.client_pid");
const std::string CLIENT_PPID("qpid.client_ppid");
-const int SESSION_FLOW_CONTROL_VER = 1;
const std::string SPACE(" ");
}
@@ -173,9 +171,6 @@ void ConnectionHandler::Handler::startOk
}
QPID_LOG(info, "Connection is a federation link");
}
- if (clientProperties.getAsInt(SESSION_FLOW_CONTROL) == SESSION_FLOW_CONTROL_VER) {
- connection.setClientThrottling();
- }
if (connection.getMgmtObject() != 0) {
string procName = clientProperties.getAsString(CLIENT_PROCESS_NAME);
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionState.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionState.h?rev=1231221&r1=1231220&r2=1231221&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionState.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionState.h Fri Jan 13 18:22:40 2012
@@ -48,7 +48,6 @@ class ConnectionState : public Connectio
heartbeatmax(120),
userProxyAuth(false), // Can proxy msgs with non-matching auth ids when true (used by federation links & clustering)
federationLink(true),
- clientSupportsThrottling(false),
clusterOrderOut(0),
isDefaultRealm(false)
{}
@@ -92,9 +91,6 @@ class ConnectionState : public Connectio
return (id == userId || (isDefaultRealm && id == userName));
}
- void setClientThrottling(bool set=true) { clientSupportsThrottling = set; }
- bool getClientThrottling() const { return clientSupportsThrottling; }
-
Broker& getBroker() { return broker; }
Broker& broker;
@@ -128,7 +124,6 @@ class ConnectionState : public Connectio
bool federationLink;
std::string federationPeerTag;
std::vector<Url> knownHosts;
- bool clientSupportsThrottling;
framing::FrameHandler* clusterOrderOut;
std::string userName;
bool isDefaultRealm;
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.cpp?rev=1231221&r1=1231220&r2=1231221&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.cpp Fri Jan 13 18:22:40 2012
@@ -24,7 +24,6 @@
#include "qpid/broker/DeliveryRecord.h"
#include "qpid/broker/SessionManager.h"
#include "qpid/broker/SessionHandler.h"
-#include "qpid/broker/RateFlowcontrol.h"
#include "qpid/sys/ClusterSafe.h"
#include "qpid/sys/Timer.h"
#include "qpid/framing/AMQContentBody.h"
@@ -62,17 +61,8 @@ SessionState::SessionState(
adapter(semanticState),
msgBuilder(&broker.getStore()),
mgmtObject(0),
- rateFlowcontrol(0),
asyncCommandCompleter(new AsyncCommandCompleter(this))
{
- uint32_t maxRate = broker.getOptions().maxSessionRate;
- if (maxRate) {
- if (handler->getConnection().getClientThrottling()) {
- rateFlowcontrol.reset(new RateFlowcontrol(maxRate));
- } else {
- QPID_LOG(warning, getId() << ": Unable to flow control client - client doesn't support");
- }
- }
if (!delayManagement) addManagementObject();
attach(h);
}
@@ -88,8 +78,6 @@ void SessionState::addManagementObject()
mgmtObject->set_attached (0);
mgmtObject->set_detachedLifespan (0);
mgmtObject->clr_expireTime();
- if (rateFlowcontrol)
- mgmtObject->set_maxClientRate(rateFlowcontrol->getRate());
agent->addObject(mgmtObject);
}
}
@@ -100,9 +88,6 @@ SessionState::~SessionState() {
semanticState.closed();
if (mgmtObject != 0)
mgmtObject->resourceDestroy ();
-
- if (flowControlTimer)
- flowControlTimer->cancel();
}
AMQP_ClientProxy& SessionState::getProxy() {
@@ -221,30 +206,6 @@ void SessionState::handleCommand(framing
}
}
-struct ScheduledCreditTask : public sys::TimerTask {
- sys::Timer& timer;
- SessionState& sessionState;
- ScheduledCreditTask(const qpid::sys::Duration& d, sys::Timer& t,
- SessionState& s) :
- TimerTask(d,"ScheduledCredit"),
- timer(t),
- sessionState(s)
- {}
-
- void fire() {
- // This is the best we can currently do to avoid a destruction/fire race
- sessionState.getConnection().requestIOProcessing(boost::bind(&ScheduledCreditTask::sendCredit, this));
- }
-
- void sendCredit() {
- if ( !sessionState.processSendCredit(0) ) {
- QPID_LOG(warning, sessionState.getId() << ": Reschedule sending credit");
- setupNextFire();
- timer.add(this);
- }
- }
-};
-
void SessionState::handleContent(AMQFrame& frame, const SequenceNumber& id)
{
if (frame.getBof() && frame.getBos()) //start of frameset
@@ -268,43 +229,6 @@ void SessionState::handleContent(AMQFram
IncompleteIngressMsgXfer xfer(this, msg);
msg->getIngressCompletion().end(xfer); // allows msg to complete xfer
}
-
- // Handle producer session flow control
- if (rateFlowcontrol && frame.getBof() && frame.getBos()) {
- if ( !processSendCredit(1) ) {
- QPID_LOG(debug, getId() << ": Schedule sending credit");
- sys::Timer& timer = getBroker().getTimer();
- // Use heuristic for scheduled credit of time for 50 messages, but not longer than 500ms
- sys::Duration d = std::min(sys::TIME_SEC * 50 / rateFlowcontrol->getRate(), 500 * sys::TIME_MSEC);
- flowControlTimer = new ScheduledCreditTask(d, timer, *this);
- timer.add(flowControlTimer);
- }
- }
-}
-
-bool SessionState::processSendCredit(uint32_t msgs)
-{
- qpid::sys::ScopedLock<Mutex> l(rateLock);
- // Check for violating flow control
- if ( msgs > 0 && rateFlowcontrol->flowStopped() ) {
- QPID_LOG(warning, getId() << ": producer throttling violation");
- // TODO: Probably do message.stop("") first time then disconnect
- // See comment on getClusterOrderProxy() in .h file
- getClusterOrderProxy().getMessage().stop("");
- return true;
- }
- AbsTime now = AbsTime::now();
- uint32_t sendCredit = rateFlowcontrol->receivedMessage(now, msgs);
- if (mgmtObject) mgmtObject->dec_clientCredit(msgs);
- if ( sendCredit>0 ) {
- QPID_LOG(debug, getId() << ": send producer credit " << sendCredit);
- getClusterOrderProxy().getMessage().flow("", 0, sendCredit);
- rateFlowcontrol->sentCredit(now, sendCredit);
- if (mgmtObject) mgmtObject->inc_clientCredit(sendCredit);
- return true;
- } else {
- return !rateFlowcontrol->flowStopped() ;
- }
}
void SessionState::sendAcceptAndCompletion()
@@ -399,17 +323,6 @@ void SessionState::readyToSend() {
QPID_LOG(debug, getId() << ": ready to send, activating output.");
assert(handler);
semanticState.attached();
- if (rateFlowcontrol) {
- qpid::sys::ScopedLock<Mutex> l(rateLock);
- // Issue initial credit - use a heuristic here issue min of 300 messages or 1 secs worth
- uint32_t credit = std::min(rateFlowcontrol->getRate(), 300U);
- QPID_LOG(debug, getId() << ": Issuing producer message credit " << credit);
- // See comment on getClusterOrderProxy() in .h file
- getClusterOrderProxy().getMessage().setFlowMode("", 0);
- getClusterOrderProxy().getMessage().flow("", 0, credit);
- rateFlowcontrol->sentCredit(AbsTime::now(), credit);
- if (mgmtObject) mgmtObject->inc_clientCredit(credit);
- }
}
Broker& SessionState::getBroker() { return broker; }
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.h?rev=1231221&r1=1231220&r2=1231221&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.h Fri Jan 13 18:22:40 2012
@@ -61,7 +61,6 @@ class ConnectionState;
class Message;
class SessionHandler;
class SessionManager;
-class RateFlowcontrol;
/**
* Broker-side session state includes session's handler chains, which
@@ -121,8 +120,6 @@ class SessionState : public qpid::Sessio
boost::intrusive_ptr<Message> getMessageInProgress() { return msgBuilder.getMessage(); }
SessionAdapter& getSessionAdapter() { return adapter; }
- bool processSendCredit(uint32_t msgs);
-
const SessionId& getSessionId() const { return getId(); }
// Used by ExecutionHandler sync command processing. Notifies
@@ -168,11 +165,6 @@ class SessionState : public qpid::Sessio
qmf::org::apache::qpid::broker::Session* mgmtObject;
qpid::framing::SequenceSet accepted;
- // State used for producer flow control (rate limited)
- qpid::sys::Mutex rateLock;
- boost::scoped_ptr<RateFlowcontrol> rateFlowcontrol;
- boost::intrusive_ptr<sys::TimerTask> flowControlTimer;
-
// sequence numbers for pending received Execution.Sync commands
std::queue<SequenceNumber> pendingExecutionSyncs;
bool currentCommandComplete;
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=1231221&r1=1231220&r2=1231221&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp Fri Jan 13 18:22:40 2012
@@ -138,8 +138,6 @@ void Connection::init() {
else { // Shadow or catch-up connection
// Passive, discard cluster-order frames
connection->setClusterOrderOutput(nullFrameHandler);
- // Disable client throttling, done by active node.
- connection->setClientThrottling(false);
}
if (!isCatchUp())
connection->setErrorListener(this);
Modified: qpid/trunk/qpid/cpp/src/tests/CMakeLists.txt
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/CMakeLists.txt?rev=1231221&r1=1231220&r2=1231221&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/CMakeLists.txt (original)
+++ qpid/trunk/qpid/cpp/src/tests/CMakeLists.txt Fri Jan 13 18:22:40 2012
@@ -133,7 +133,6 @@ set(unit_tests_to_build
QueueEvents
ProxyTest
RetryList
- RateFlowcontrolTest
FrameDecoder
ReplicationTest
ClientMessageTest
Modified: qpid/trunk/qpid/cpp/src/tests/Makefile.am
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/Makefile.am?rev=1231221&r1=1231220&r2=1231221&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/Makefile.am (original)
+++ qpid/trunk/qpid/cpp/src/tests/Makefile.am Fri Jan 13 18:22:40 2012
@@ -116,7 +116,6 @@ unit_test_SOURCES= unit_test.cpp unit_te
QueueEvents.cpp \
ProxyTest.cpp \
RetryList.cpp \
- RateFlowcontrolTest.cpp \
FrameDecoder.cpp \
ReplicationTest.cpp \
ClientMessageTest.cpp \
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org