diff options
Diffstat (limited to 'cpp/src/qpid/cluster/Connection.cpp')
| -rw-r--r-- | cpp/src/qpid/cluster/Connection.cpp | 235 |
1 files changed, 180 insertions, 55 deletions
diff --git a/cpp/src/qpid/cluster/Connection.cpp b/cpp/src/qpid/cluster/Connection.cpp index e9b718e6de..394749aad2 100644 --- a/cpp/src/qpid/cluster/Connection.cpp +++ b/cpp/src/qpid/cluster/Connection.cpp @@ -7,9 +7,9 @@ * 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 @@ -24,6 +24,8 @@ #include "Cluster.h" #include "UpdateReceiver.h" #include "qpid/assert.h" +#include "qpid/broker/DtxAck.h" +#include "qpid/broker/DtxBuffer.h" #include "qpid/broker/SessionState.h" #include "qpid/broker/SemanticState.h" #include "qpid/broker/TxBuffer.h" @@ -35,6 +37,7 @@ #include "qpid/broker/Fairshare.h" #include "qpid/broker/Link.h" #include "qpid/broker/Bridge.h" +#include "qpid/broker/StatefulQueueObserver.h" #include "qpid/broker/Queue.h" #include "qpid/framing/enum.h" #include "qpid/framing/AMQFrame.h" @@ -78,7 +81,7 @@ const std::string shadowPrefix("[shadow]"); Connection::Connection(Cluster& c, sys::ConnectionOutputHandler& out, const std::string& mgmtId, const ConnectionId& id, const qpid::sys::SecuritySettings& external) - : cluster(c), self(id), catchUp(false), output(*this, out), + : cluster(c), self(id), catchUp(false), announced(false), output(*this, out), connectionCtor(&output, cluster.getBroker(), mgmtId, external, false, 0, true), expectProtocolHeader(false), mcastFrameHandler(cluster.getMulticast(), self), @@ -90,13 +93,15 @@ Connection::Connection(Cluster& c, sys::ConnectionOutputHandler& out, Connection::Connection(Cluster& c, sys::ConnectionOutputHandler& out, const std::string& mgmtId, MemberId member, bool isCatchUp, bool isLink, const qpid::sys::SecuritySettings& external -) : cluster(c), self(member, ++idCounter), catchUp(isCatchUp), output(*this, out), +) : cluster(c), self(member, ++idCounter), catchUp(isCatchUp), announced(false), output(*this, out), connectionCtor(&output, cluster.getBroker(), mgmtId, external, isLink, isCatchUp ? ++catchUpId : 0, - isCatchUp), // isCatchUp => shadow + // The first catch-up connection is not considered a shadow + // as it needs to be authenticated. + isCatchUp && self.second > 1), expectProtocolHeader(isLink), mcastFrameHandler(cluster.getMulticast(), self), updateIn(c.getUpdateReceiver()), @@ -113,7 +118,7 @@ Connection::Connection(Cluster& c, sys::ConnectionOutputHandler& out, if (!updateIn.nextShadowMgmtId.empty()) connectionCtor.mgmtId = updateIn.nextShadowMgmtId; updateIn.nextShadowMgmtId.clear(); - } + } init(); QPID_LOG(debug, cluster << " local connection " << *this); } @@ -143,7 +148,7 @@ void Connection::init() { // Called when we have consumed a read buffer to give credit to the // connection layer to continue reading. void Connection::giveReadCredit(int credit) { - if (cluster.getSettings().readMax && credit) + if (cluster.getSettings().readMax && credit) output.giveReadCredit(credit); } @@ -166,7 +171,7 @@ void Connection::announce( AMQFrame frame; while (frame.decode(buf)) connection->received(frame); - connection->setUserId(username); + connection->setUserId(username); } // Do managment actions now that the connection is replicated. connection->raiseConnectEvent(); @@ -193,7 +198,7 @@ void Connection::received(framing::AMQFrame& f) { << *this << ": " << f); return; } - QPID_LOG(trace, cluster << " RECV " << *this << ": " << f); + QPID_LOG_IF(trace, Cluster::loggable(f), cluster << " RECV " << *this << ": " << f); if (isLocal()) { // Local catch-up connection. currentChannel = f.getChannel(); if (!framing::invoke(*this, *f.getBody()).wasHandled()) @@ -201,7 +206,7 @@ void Connection::received(framing::AMQFrame& f) { } else { // Shadow or updated catch-up connection. if (f.getMethod() && f.getMethod()->isA<ConnectionCloseBody>()) { - if (isShadow()) + if (isShadow()) cluster.addShadowConnection(this); AMQFrame ok((ConnectionCloseOkBody())); connection->getOutput().send(ok); @@ -213,16 +218,9 @@ void Connection::received(framing::AMQFrame& f) { } } -bool Connection::checkUnsupported(const AMQBody& body) { - std::string message; - if (body.getMethod()) { - switch (body.getMethod()->amqpClassId()) { - case DTX_CLASS_ID: message = "DTX transactions are not currently supported by cluster."; break; - } - } - if (!message.empty()) - connection->close(connection::CLOSE_CODE_FRAMING_ERROR, message); - return !message.empty(); +bool Connection::checkUnsupported(const AMQBody&) { + // Throw an exception for unsupported commands. Currently all are supported. + return false; } struct GiveReadCreditOnExit { @@ -241,7 +239,7 @@ void Connection::deliverDoOutput(uint32_t limit) { void Connection::deliveredFrame(const EventFrame& f) { GiveReadCreditOnExit gc(*this, f.readCredit); assert(!catchUp); - currentChannel = f.frame.getChannel(); + currentChannel = f.frame.getChannel(); if (f.frame.getBody() // frame can be emtpy with just readCredit && !framing::invoke(*this, *f.frame.getBody()).wasHandled() // Connection contol. && !checkUnsupported(*f.frame.getBody())) // Unsupported operation. @@ -255,7 +253,7 @@ void Connection::deliveredFrame(const EventFrame& f) { } } -// A local connection is closed by the network layer. +// A local connection is closed by the network layer. Called in the connection thread. void Connection::closed() { try { if (isUpdated()) { @@ -272,8 +270,9 @@ void Connection::closed() { // closed and process any outstanding frames from the cluster // until self-delivery of deliver-close. output.closeOutput(); - cluster.getMulticast().mcastControl( - ClusterConnectionDeliverCloseBody(), self); + if (announced) + cluster.getMulticast().mcastControl( + ClusterConnectionDeliverCloseBody(), self); } } catch (const std::exception& e) { @@ -287,7 +286,7 @@ void Connection::deliverClose () { cluster.erase(self); } -// Close the connection +// Close the connection void Connection::close() { if (connection.get()) { QPID_LOG(debug, cluster << " closed connection " << *this); @@ -320,10 +319,10 @@ size_t Connection::decode(const char* data, size_t size) { while (localDecoder.decode(buf)) received(localDecoder.getFrame()); if (!wasOpen && connection->isOpen()) { - // Connections marked as federation links are allowed to proxy + // Connections marked with setUserProxyAuth are allowed to proxy // messages with user-ID that doesn't match the connection's // authenticated ID. This is important for updates. - connection->setFederationLink(isCatchUp()); + connection->setUserProxyAuth(isCatchUp()); } } else { // Multicast local connections. @@ -332,9 +331,9 @@ size_t Connection::decode(const char* data, size_t size) { if (!checkProtocolHeader(ptr, size)) // Updates ptr return 0; // Incomplete header - if (!connection->isOpen()) + if (!connection->isOpen()) processInitialFrames(ptr, end-ptr); // Updates ptr - + if (connection->isOpen() && end - ptr > 0) { // We're multi-casting, we will give read credit on delivery. grc.credit = 0; @@ -384,6 +383,7 @@ void Connection::processInitialFrames(const char*& ptr, size_t size) { connection->getUserId(), initialFrames), getId()); + announced = true; initialFrames.clear(); } } @@ -406,11 +406,11 @@ void Connection::shadowSetUser(const std::string& userId) { void Connection::consumerState(const string& name, bool blocked, bool notifyEnabled, const SequenceNumber& position) { - broker::SemanticState::ConsumerImpl& c = semanticState().find(name); - c.position = position; - c.setBlocked(blocked); - if (notifyEnabled) c.enableNotify(); else c.disableNotify(); - updateIn.consumerNumbering.add(c.shared_from_this()); + broker::SemanticState::ConsumerImpl::shared_ptr c = semanticState().find(name); + c->position = position; + c->setBlocked(blocked); + if (notifyEnabled) c->enableNotify(); else c->disableNotify(); + updateIn.consumerNumbering.add(c); } @@ -421,7 +421,8 @@ void Connection::sessionState( const SequenceNumber& expected, const SequenceNumber& received, const SequenceSet& unknownCompleted, - const SequenceSet& receivedIncomplete) + const SequenceSet& receivedIncomplete, + bool dtxSelected) { sessionState().setState( replayStart, @@ -431,8 +432,10 @@ void Connection::sessionState( received, unknownCompleted, receivedIncomplete); - QPID_LOG(debug, cluster << " received session state update for " << sessionState().getId()); - // The output tasks will be added later in the update process. + if (dtxSelected) semanticState().selectDtx(); + QPID_LOG(debug, cluster << " received session state update for " + << sessionState().getId()); + // The output tasks will be added later in the update process. connection->getOutputTasks().removeAll(); } @@ -441,7 +444,7 @@ void Connection::outputTask(uint16_t channel, const std::string& name) { if (!session) throw Exception(QPID_MSG(cluster << " channel not attached " << *this << "[" << channel << "] ")); - OutputTask* task = &session->getSemanticState().find(name); + OutputTask* task = session->getSemanticState().find(name).get(); connection->getOutputTasks().addOutputTask(task); } @@ -461,11 +464,24 @@ void Connection::shadowReady( output.setSendMax(sendMax); } +void Connection::setDtxBuffer(const UpdateReceiver::DtxBufferRef& bufRef) { + broker::DtxManager& mgr = cluster.getBroker().getDtxManager(); + broker::DtxWorkRecord* record = mgr.getWork(bufRef.xid); + broker::DtxBuffer::shared_ptr buffer = (*record)[bufRef.index]; + if (bufRef.suspended) + bufRef.semanticState->getSuspendedXids()[bufRef.xid] = buffer; + else + bufRef.semanticState->setDtxBuffer(buffer); +} + +// Marks the end of the update. void Connection::membership(const FieldTable& joiners, const FieldTable& members, const framing::SequenceNumber& frameSeq) { QPID_LOG(debug, cluster << " incoming update complete on connection " << *this); updateIn.consumerNumbering.clear(); + for_each(updateIn.dtxBuffers.begin(), updateIn.dtxBuffers.end(), + boost::bind(&Connection::setDtxBuffer, this, _1)); closeUpdated(); cluster.updateInDone(ClusterMap(joiners, members, frameSeq)); } @@ -478,7 +494,7 @@ void Connection::retractOffer() { void Connection::closeUpdated() { self.second = 0; // Mark this as completed update connection. - if (connection.get()) + if (connection.get()) connection->close(connection::CLOSE_CODE_NORMAL, "OK"); } @@ -529,12 +545,20 @@ void Connection::deliveryRecord(const string& qname, m = getUpdateMessage(); m.queue = queue.get(); m.position = position; - if (enqueued) queue->updateEnqueued(m); //inform queue of the message + if (enqueued) queue->updateEnqueued(m); //inform queue of the message } else { // Message at original position in original queue - m = queue->find(position); + queue->find(position, m); } - if (!m.payload) - throw Exception(QPID_MSG("deliveryRecord no update message")); + // FIXME aconway 2011-08-19: removed: + // if (!m.payload) + // throw Exception(QPID_MSG("deliveryRecord no update message")); + // + // It seems this could happen legitimately in the case one + // session browses message M, then another session acquires + // it. In that case the browsers delivery record is !acquired + // but the message is not on its original Queue. In that case + // we'll get a deliveryRecord with no payload for the browser. + // } broker::DeliveryRecord dr(m, queue, tag, acquired, accepted, windowing, credit); @@ -542,7 +566,11 @@ void Connection::deliveryRecord(const string& qname, if (cancelled) dr.cancel(dr.getTag()); if (completed) dr.complete(); if (ended) dr.setEnded(); // Exsitance of message - semanticState().record(dr); // Part of the session's unacked list. + + if (dtxBuffer) // Record for next dtx-ack + dtxAckRecords.push_back(dr); + else + semanticState().record(dr); // Record on session's unacked list. } void Connection::queuePosition(const string& qname, const SequenceNumber& position) { @@ -556,8 +584,46 @@ void Connection::queueFairshareState(const std::string& qname, const uint8_t pri } } -void Connection::expiryId(uint64_t id) { - cluster.getExpiryPolicy().setId(id); + +namespace { +// find a StatefulQueueObserver that matches a given identifier +class ObserverFinder { + const std::string id; + boost::shared_ptr<broker::QueueObserver> target; + ObserverFinder(const ObserverFinder&) {} + public: + ObserverFinder(const std::string& _id) : id(_id) {} + broker::StatefulQueueObserver *getObserver() + { + if (target) + return dynamic_cast<broker::StatefulQueueObserver *>(target.get()); + return 0; + } + void operator() (boost::shared_ptr<broker::QueueObserver> o) + { + if (!target) { + broker::StatefulQueueObserver *p = dynamic_cast<broker::StatefulQueueObserver *>(o.get()); + if (p && p->getId() == id) { + target = o; + } + } + } +}; +} + + +void Connection::queueObserverState(const std::string& qname, const std::string& observerId, const FieldTable& state) +{ + boost::shared_ptr<broker::Queue> queue(findQueue(qname)); + ObserverFinder finder(observerId); // find this observer + queue->eachObserver<ObserverFinder &>(finder); + broker::StatefulQueueObserver *so = finder.getObserver(); + if (so) { + so->setState( state ); + QPID_LOG(debug, "updated queue observer " << observerId << "'s state on queue " << qname << "; ..."); + return; + } + QPID_LOG(error, "Failed to find observer " << observerId << " state on queue " << qname << "; this will result in inconsistencies."); } std::ostream& operator<<(std::ostream& o, const Connection& c) { @@ -574,6 +640,7 @@ std::ostream& operator<<(std::ostream& o, const Connection& c) { void Connection::txStart() { txBuffer.reset(new broker::TxBuffer()); } + void Connection::txAccept(const framing::SequenceSet& acked) { txBuffer->enlist(boost::shared_ptr<broker::TxAccept>( new broker::TxAccept(acked, semanticState().getUnacked()))); @@ -589,9 +656,11 @@ void Connection::txEnqueue(const std::string& queue) { new broker::RecoveredEnqueue(findQueue(queue), getUpdateMessage().payload))); } -void Connection::txPublish(const framing::Array& queues, bool delivered) { - boost::shared_ptr<broker::TxPublish> txPub(new broker::TxPublish(getUpdateMessage().payload)); - for (framing::Array::const_iterator i = queues.begin(); i != queues.end(); ++i) +void Connection::txPublish(const framing::Array& queues, bool delivered) +{ + boost::shared_ptr<broker::TxPublish> txPub( + new broker::TxPublish(getUpdateMessage().payload)); + for (framing::Array::const_iterator i = queues.begin(); i != queues.end(); ++i) txPub->deliverTo(findQueue((*i)->get<std::string>())); txPub->delivered = delivered; txBuffer->enlist(txPub); @@ -605,6 +674,51 @@ void Connection::accumulatedAck(const qpid::framing::SequenceSet& s) { semanticState().setAccumulatedAck(s); } +void Connection::dtxStart(const std::string& xid, + bool ended, + bool suspended, + bool failed, + bool expired) +{ + dtxBuffer.reset(new broker::DtxBuffer(xid, ended, suspended, failed, expired)); + txBuffer = dtxBuffer; +} + +void Connection::dtxEnd() { + broker::DtxManager& mgr = cluster.getBroker().getDtxManager(); + std::string xid = dtxBuffer->getXid(); + if (mgr.exists(xid)) + mgr.join(xid, dtxBuffer); + else + mgr.start(xid, dtxBuffer); + dtxBuffer.reset(); + txBuffer.reset(); +} + +// Sent after all DeliveryRecords for a dtx-ack have been collected in dtxAckRecords +void Connection::dtxAck() { + dtxBuffer->enlist( + boost::shared_ptr<broker::DtxAck>(new broker::DtxAck(dtxAckRecords))); + dtxAckRecords.clear(); +} + +void Connection::dtxBufferRef(const std::string& xid, uint32_t index, bool suspended) { + // Save the association between DtxBuffers and the session so we + // can set the DtxBuffers at the end of the update when the + // DtxManager has been replicated. + updateIn.dtxBuffers.push_back( + UpdateReceiver::DtxBufferRef(xid, index, suspended, &semanticState())); +} + +// Sent at end of work record. +void Connection::dtxWorkRecord(const std::string& xid, bool prepared, uint32_t timeout) +{ + broker::DtxManager& mgr = cluster.getBroker().getDtxManager(); + if (timeout) mgr.setTimeout(xid, timeout); + if (prepared) mgr.prepare(xid); +} + + void Connection::exchange(const std::string& encoded) { Buffer buf(const_cast<char*>(encoded.data()), encoded.size()); broker::Exchange::shared_ptr ex = broker::Exchange::decode(cluster.getBroker().getExchanges(), buf); @@ -614,12 +728,6 @@ void Connection::exchange(const std::string& encoded) { QPID_LOG(debug, cluster << " updated exchange " << ex->getName()); } -void Connection::queue(const std::string& encoded) { - Buffer buf(const_cast<char*>(encoded.data()), encoded.size()); - broker::Queue::shared_ptr q = broker::Queue::decode(cluster.getBroker().getQueues(), buf); - QPID_LOG(debug, cluster << " updated queue " << q->getName()); -} - void Connection::sessionError(uint16_t , const std::string& msg) { // Ignore errors before isOpen(), we're not multicasting yet. if (connection->isOpen()) @@ -678,6 +786,23 @@ void Connection::config(const std::string& encoded) { else throw Exception(QPID_MSG("Update failed, invalid kind of config: " << kind)); } +void Connection::doCatchupIoCallbacks() { + // We need to process IO callbacks during the catch-up phase in + // order to service asynchronous completions for messages + // transferred during catch-up. + + if (catchUp) getBrokerConnection()->doIoCallbacks(); +} + +void Connection::clock(uint64_t time) { + QPID_LOG(debug, "Cluster connection received time update"); + cluster.clock(time); +} + +void Connection::queueDequeueSincePurgeState(const std::string& qname, uint32_t dequeueSincePurge) { + boost::shared_ptr<broker::Queue> queue(findQueue(qname)); + queue->setDequeueSincePurge(dequeueSincePurge); +} }} // Namespace qpid::cluster |