Author: aconway
Date: Sun Dec 7 18:18:03 2008
New Revision: 724233
URL: http://svn.apache.org/viewvc?rev=724233&view=rev
Log:
OutputControl and subclasses: added giveReadCredit() for IO level flow control.
Cluster: Set read credit limit for cluster connections.
Modified:
incubator/qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/Connection.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/Connection.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.h
incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connector.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/NoOpConnectionOutputHandler.h
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.h
incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AggregateOutput.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AggregateOutput.h
incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIO.h
incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOHandler.h
incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ConnectionOutputHandlerPtr.h
incubator/qpid/trunk/qpid/cpp/src/qpid/sys/OutputControl.h
incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ssl/SslHandler.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ssl/SslHandler.h
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/Connection.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/Connection.cpp?rev=724233&r1=724232&r2=724233&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/Connection.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/Connection.cpp Sun Dec 7
18:18:03 2008
@@ -92,6 +92,7 @@
}
void Connection::activateOutput() { output.activateOutput(); }
+void Connection::giveReadCredit(int32_t credit) {
output.giveReadCredit(credit); }
void Connection::close() {
// Close the output queue.
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/Connection.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/Connection.h?rev=724233&r1=724232&r2=724233&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/Connection.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/Connection.h Sun Dec 7
18:18:03 2008
@@ -62,6 +62,7 @@
bool isClosed() const;
bool canEncode();
void activateOutput();
+ void giveReadCredit(int32_t);
void closed(); // connection closed by peer.
void close(); // closing from this end.
void send(framing::AMQFrame&);
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.cpp?rev=724233&r1=724232&r2=724233&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.cpp Sun Dec 7
18:18:03 2008
@@ -121,6 +121,11 @@
getConnection().outputTasks.activateOutput();
}
+void SessionState::giveReadCredit(int32_t credit) {
+ if (isAttached())
+ getConnection().outputTasks.giveReadCredit(credit);
+}
+
ManagementObject* SessionState::GetManagementObject (void) const
{
return (ManagementObject*) mgmtObject;
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.h?rev=724233&r1=724232&r2=724233&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.h Sun Dec 7
18:18:03 2008
@@ -86,6 +86,7 @@
/** OutputControl **/
void activateOutput();
+ void giveReadCredit(int32_t);
void senderCompleted(const framing::SequenceSet& ranges);
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connector.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connector.cpp?rev=724233&r1=724232&r2=724233&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connector.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connector.cpp Sun Dec 7
18:18:03 2008
@@ -134,7 +134,7 @@
void handleClosed();
bool closeInternal();
- void readbuff(qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*);
+ bool readbuff(qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*);
void writebuff(qpid::sys::AsynchIO&);
void writeDataBlock(const framing::AMQDataBlock& data);
void eof(qpid::sys::AsynchIO&);
@@ -340,7 +340,7 @@
if (encode.getPosition() > 0) writeOne();
}
-void TCPConnector::readbuff(AsynchIO& aio, AsynchIO::BufferBase* buff) {
+bool TCPConnector::readbuff(AsynchIO& aio, AsynchIO::BufferBase* buff) {
framing::Buffer in(buff->bytes+buff->dataStart, buff->dataCount);
if (!initiated) {
@@ -367,6 +367,7 @@
// Give whole buffer back to aio subsystem
aio.queueReadBuffer(buff);
}
+ return true;
}
void TCPConnector::writebuff(AsynchIO& aio_) {
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp?rev=724233&r1=724232&r2=724233&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp Sun Dec 7
18:18:03 2008
@@ -62,17 +62,21 @@
const std::string& wrappedId, ConnectionId myId)
: cluster(c), self(myId), catchUp(false), output(*this, out),
connection(&output, cluster.getBroker(), wrappedId)
-{
- QPID_LOG(debug, cluster << " new connection: " << *this);
-}
+{ init(); }
// Local connections
Connection::Connection(Cluster& c, sys::ConnectionOutputHandler& out,
const std::string& wrappedId, MemberId myId, bool
isCatchUp)
: cluster(c), self(myId, this), catchUp(isCatchUp), output(*this, out),
connection(&output, cluster.getBroker(), wrappedId)
-{
+{ init(); }
+
+void Connection::init() {
QPID_LOG(debug, cluster << " new connection: " << *this);
+ if (isLocal() && !isCatchUp()) {
+ // FIXME aconway 2008-12-05: configurable credit limit
+ output.giveReadCredit(3);
+ }
}
Connection::~Connection() {
@@ -187,6 +191,7 @@
Buffer buf(const_cast<char*>(buffer), size);
while (localDecoder.decode(buf))
received(localDecoder.frame);
+ output.giveReadCredit(1);
}
else { // Multicast local connections.
assert(isLocal());
@@ -200,6 +205,8 @@
++deliverSeq;
while (mcastDecoder.decode(buf))
delivered(mcastDecoder.frame);
+ if (isLocal())
+ output.giveReadCredit(1);
}
broker::SessionState& Connection::sessionState() {
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h?rev=724233&r1=724232&r2=724233&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h Sun Dec 7
18:18:03 2008
@@ -142,6 +142,7 @@
void exchange(const std::string& encoded);
private:
+ void init();
bool checkUnsupported(const framing::AMQBody& body);
void deliverClose();
void deliverDoOutput(uint32_t requested);
@@ -167,6 +168,8 @@
framing::ChannelId currentChannel;
boost::shared_ptr<broker::TxBuffer> txBuffer;
+ int FIXMEcredit; // FIXME aconway 2008-12-05: remove
+
friend std::ostream& operator<<(std::ostream&, const Connection&);
};
Modified:
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/NoOpConnectionOutputHandler.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/NoOpConnectionOutputHandler.h?rev=724233&r1=724232&r2=724233&view=diff
==============================================================================
---
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/NoOpConnectionOutputHandler.h
(original)
+++
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/NoOpConnectionOutputHandler.h
Sun Dec 7 18:18:03 2008
@@ -39,6 +39,7 @@
virtual void send(framing::AMQFrame&) {}
virtual void close() {}
virtual void activateOutput() {}
+ virtual void giveReadCredit(int32_t) {}
};
}} // namespace qpid::cluster
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.cpp?rev=724233&r1=724232&r2=724233&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.cpp
(original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.cpp Sun
Dec 7 18:18:03 2008
@@ -55,6 +55,8 @@
}
}
+void OutputInterceptor::giveReadCredit(int32_t credit) {
next->giveReadCredit(credit); }
+
// Called in write thread when the IO layer has no more data to write.
// We do nothing in the write thread, we run doOutput only on delivery
// of doOutput requests.
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.h?rev=724233&r1=724232&r2=724233&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.h
(original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.h Sun Dec
7 18:18:03 2008
@@ -43,6 +43,7 @@
// sys::ConnectionOutputHandler functions
void send(framing::AMQFrame& f);
void activateOutput();
+ void giveReadCredit(int32_t);
void close();
size_t getBuffered() const;
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AggregateOutput.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AggregateOutput.cpp?rev=724233&r1=724232&r2=724233&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AggregateOutput.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AggregateOutput.cpp Sun Dec 7
18:18:03 2008
@@ -26,10 +26,9 @@
namespace qpid {
namespace sys {
-void AggregateOutput::activateOutput()
-{
- control.activateOutput();
-}
+void AggregateOutput::activateOutput() { control.activateOutput(); }
+
+void AggregateOutput::giveReadCredit(int32_t credit) {
control.giveReadCredit(credit); }
bool AggregateOutput::hasOutput() {
for (TaskList::const_iterator i = tasks.begin(); i != tasks.end(); ++i)
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AggregateOutput.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AggregateOutput.h?rev=724233&r1=724232&r2=724233&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AggregateOutput.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AggregateOutput.h Sun Dec 7
18:18:03 2008
@@ -43,6 +43,8 @@
AggregateOutput(OutputControl& c) : next(0), control(c) {};
//this may be called on any thread
void activateOutput();
+ void giveReadCredit(int32_t);
+
//all the following will be called on the same thread
bool doOutput();
bool hasOutput();
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIO.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIO.h?rev=724233&r1=724232&r2=724233&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIO.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIO.h Sun Dec 7 18:18:03
2008
@@ -108,7 +108,7 @@
public:
typedef AsynchIOBufferBase BufferBase;
- typedef boost::function2<void, AsynchIO&, BufferBase*> ReadCallback;
+ typedef boost::function2<bool, AsynchIO&, BufferBase*> ReadCallback;
typedef boost::function1<void, AsynchIO&> EofCallback;
typedef boost::function1<void, AsynchIO&> DisconnectCallback;
typedef boost::function2<void, AsynchIO&, const Socket&> ClosedCallback;
@@ -137,6 +137,7 @@
virtual void notifyPendingWrite() = 0;
virtual void queueWriteClose() = 0;
virtual bool writeQueueEmpty() = 0;
+ virtual void startReading() = 0;
virtual BufferBase* getQueuedBuffer() = 0;
protected:
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp?rev=724233&r1=724232&r2=724233&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp Sun Dec 7
18:18:03 2008
@@ -44,7 +44,8 @@
factory(f),
codec(0),
readError(false),
- isClient(false)
+ isClient(false),
+ readCredit(InfiniteCredit)
{}
AsynchIOHandler::~AsynchIOHandler() {
@@ -79,9 +80,22 @@
}
// Input side
-void AsynchIOHandler::readbuff(AsynchIO& , AsynchIO::BufferBase* buff) {
- if (readError) {
+void AsynchIOHandler::giveReadCredit(int32_t credit) {
+ // Check whether we started in the don't about credit state
+ if (readCredit.boolCompareAndSwap(InfiniteCredit, credit))
+ return;
+ else if (readCredit.fetchAndAdd(credit) != 0)
return;
+ // Lock and retest credit to make sure we don't race with decreasing credit
+ ScopedLock<Mutex> l(creditLock);
+ assert(readCredit.get() >= 0);
+ if (readCredit.get() != 0)
+ aio->startReading();
+}
+
+bool AsynchIOHandler::readbuff(AsynchIO& , AsynchIO::BufferBase* buff) {
+ if (readError) {
+ return false;
}
size_t decoded = 0;
if (codec) { // Already initiated
@@ -125,6 +139,17 @@
// Give whole buffer back to aio subsystem
aio->queueReadBuffer(buff);
}
+ // Check here for read credit
+ if (readCredit.get() != InfiniteCredit) {
+ if (--readCredit == 0) {
+ // Lock and retest credit to make sure we don't race with
increasing credit
+ ScopedLock<Mutex> l(creditLock);
+ assert(readCredit.get() >= 0);
+ if (readCredit.get() == 0)
+ return false;
+ }
+ }
+ return true;
}
void AsynchIOHandler::eof(AsynchIO&) {
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOHandler.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOHandler.h?rev=724233&r1=724232&r2=724233&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOHandler.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOHandler.h Sun Dec 7
18:18:03 2008
@@ -23,6 +23,8 @@
#include "OutputControl.h"
#include "ConnectionCodec.h"
+#include "AtomicValue.h"
+#include "Mutex.h"
namespace qpid {
@@ -43,6 +45,9 @@
ConnectionCodec* codec;
bool readError;
bool isClient;
+ AtomicValue<int32_t> readCredit;
+ static const int32_t InfiniteCredit = -1;
+ Mutex creditLock;
void write(const framing::ProtocolInitiation&);
@@ -56,9 +61,10 @@
// Output side
void close();
void activateOutput();
+ void giveReadCredit(int32_t credit);
// Input side
- void readbuff(AsynchIO& aio, AsynchIOBufferBase* buff);
+ bool readbuff(AsynchIO& aio, AsynchIOBufferBase* buff);
void eof(AsynchIO& aio);
void disconnect(AsynchIO& aio);
Modified:
incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ConnectionOutputHandlerPtr.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ConnectionOutputHandlerPtr.h?rev=724233&r1=724232&r2=724233&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ConnectionOutputHandlerPtr.h
(original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ConnectionOutputHandlerPtr.h Sun
Dec 7 18:18:03 2008
@@ -44,6 +44,7 @@
void close() { next->close(); }
size_t getBuffered() const { return next->getBuffered(); }
void activateOutput() { next->activateOutput(); }
+ void giveReadCredit(int32_t credit) { next->giveReadCredit(credit); }
void send(framing::AMQFrame& f) { next->send(f); }
private:
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/OutputControl.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/OutputControl.h?rev=724233&r1=724232&r2=724233&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/OutputControl.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/OutputControl.h Sun Dec 7
18:18:03 2008
@@ -18,6 +18,9 @@
* under the License.
*
*/
+
+#include "IntegerTypes.h"
+
#ifndef _OutputControl_
#define _OutputControl_
@@ -29,6 +32,7 @@
public:
virtual ~OutputControl() {}
virtual void activateOutput() = 0;
+ virtual void giveReadCredit(int32_t credit) = 0;
};
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp?rev=724233&r1=724232&r2=724233&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp Sun Dec 7
18:18:03 2008
@@ -265,6 +265,7 @@
virtual void notifyPendingWrite();
virtual void queueWriteClose();
virtual bool writeQueueEmpty();
+ virtual void startReading();
virtual BufferBase* getQueuedBuffer();
private:
@@ -381,6 +382,10 @@
return writeQueue.empty();
}
+void AsynchIO::startReading() {
+ DispatchHandle::rewatchRead();
+}
+
/** Return a queued buffer if there are enough
* to spare
*/
@@ -418,7 +423,12 @@
threadReadTotal += rc;
readTotal += rc;
- readCallback(*this, buff);
+ if (!readCallback(*this, buff)) {
+ // We were told to flow control reading at this point
+ h.unwatchRead();
+ break;
+ }
+
if (rc != readCount) {
// If we didn't fill the read buffer then time to stop
reading
break;
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ssl/SslHandler.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ssl/SslHandler.cpp?rev=724233&r1=724232&r2=724233&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ssl/SslHandler.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ssl/SslHandler.cpp Sun Dec 7
18:18:03 2008
@@ -80,6 +80,10 @@
aio->notifyPendingWrite();
}
+void SslHandler::giveReadCredit(int32_t) {
+ // FIXME aconway 2008-12-05: not yet implemented.
+}
+
// Input side
void SslHandler::readbuff(SslIO& , SslIO::BufferBase* buff) {
if (readError) {
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ssl/SslHandler.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ssl/SslHandler.h?rev=724233&r1=724232&r2=724233&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ssl/SslHandler.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ssl/SslHandler.h Sun Dec 7
18:18:03 2008
@@ -58,6 +58,7 @@
// Output side
void close();
void activateOutput();
+ void giveReadCredit(int32_t);
// Input side
void readbuff(SslIO& aio, SslIOBufferBase* buff);
