You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gs...@apache.org on 2015/10/26 20:32:42 UTC
svn commit: r1710683 - in /qpid/trunk/qpid/cpp/src/qpid/broker/amqp:
Incoming.cpp Session.cpp Session.h
Author: gsim
Date: Mon Oct 26 19:32:42 2015
New Revision: 1710683
URL: http://svn.apache.org/viewvc?rev=1710683&view=rev
Log:
QPID-6790: ensure that asynchronous completion of incoming transfers does not result in trying to settle deliveries implicitly settled by the freeing of their link
Modified:
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Incoming.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.h
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Incoming.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Incoming.cpp?rev=1710683&r1=1710682&r2=1710683&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Incoming.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Incoming.cpp Mon Oct 26 19:32:42 2015
@@ -150,6 +150,7 @@ void DecodingIncoming::deliver(boost::in
received->begin();
handle(message, session.getTransaction(delivery));
Transfer t(delivery, sessionPtr);
+ sessionPtr->pending_accept(delivery);
received->end(t);
}
}}} // namespace qpid::broker::amqp
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp?rev=1710683&r1=1710682&r2=1710683&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp Mon Oct 26 19:32:42 2015
@@ -607,6 +607,7 @@ void Session::detach(pn_link_t* link, bo
} else {
IncomingLinks::iterator i = incoming.find(link);
if (i != incoming.end()) {
+ abort_pending(link);
i->second->detached(closed);
incoming.erase(i);
QPID_LOG(debug, "Incoming link detached");
@@ -614,17 +615,51 @@ void Session::detach(pn_link_t* link, bo
}
}
+void Session::pending_accept(pn_delivery_t* delivery)
+{
+ qpid::sys::Mutex::ScopedLock l(lock);
+ pending.insert(delivery);
+}
+
+bool Session::clear_pending(pn_delivery_t* delivery)
+{
+ qpid::sys::Mutex::ScopedLock l(lock);
+ std::set<pn_delivery_t*>::iterator i = pending.find(delivery);
+ if (i != pending.end()) {
+ pending.erase(i);
+ return true;
+ } else {
+ return false;
+ }
+}
+
+void Session::abort_pending(pn_link_t* link)
+{
+ qpid::sys::Mutex::ScopedLock l(lock);
+ for (std::set<pn_delivery_t*>::iterator i = pending.begin(); i != pending.end();) {
+ if (pn_delivery_link(*i) == link) {
+ pn_delivery_settle(*i);
+ pending.erase(i++);
+ } else {
+ ++i;
+ }
+ }
+}
+
void Session::accepted(pn_delivery_t* delivery, bool sync)
{
if (sync) {
- //this is on IO thread
- pn_delivery_update(delivery, PN_ACCEPTED);
- pn_delivery_settle(delivery);//do we need to check settlement modes/orders?
- incomingMessageAccepted();
+ if (clear_pending(delivery))
+ {
+ //this is on IO thread
+ pn_delivery_update(delivery, PN_ACCEPTED);
+ pn_delivery_settle(delivery);//do we need to check settlement modes/orders?
+ incomingMessageAccepted();
+ }
} else {
//this is not on IO thread, need to delay processing until on IO thread
qpid::sys::Mutex::ScopedLock l(lock);
- if (!deleted) {
+ if (!deleted && pending.find(delivery) != pending.end()) {
completed.push_back(delivery);
out.activateOutput();
}
@@ -926,6 +961,7 @@ void IncomingToCoordinator::deliver(boos
if (i != args.end()) {
std::string id = *i;
bool failed = ++i != args.end() ? i->asBool() : false;
+ session.pending_accept(delivery);
session.discharge(id, failed, delivery);
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.h?rev=1710683&r1=1710682&r2=1710683&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.h Mon Oct 26 19:32:42 2015
@@ -77,8 +77,10 @@ class Session : public ManagedSession, p
*/
void attach(pn_link_t* link, const std::string& src, const std::string& tgt, boost::shared_ptr<Relay>);
- //called when a transfer is completly processed (e.g.including stored on disk)
+ //called when a transfer is completely processed (e.g.including stored on disk)
void accepted(pn_delivery_t*, bool sync);
+ //called to indicate that the delivery will be accepted asynchronously
+ void pending_accept(pn_delivery_t*);
//called when async transaction completes
void committed(bool sync);
@@ -104,6 +106,7 @@ class Session : public ManagedSession, p
IncomingLinks incoming;
OutgoingLinks outgoing;
std::deque<pn_delivery_t*> completed;
+ std::set<pn_delivery_t*> pending;
bool deleted;
qpid::sys::Mutex lock;
std::set< boost::shared_ptr<Queue> > exclusiveQueues;
@@ -138,6 +141,8 @@ class Session : public ManagedSession, p
void setupIncoming(pn_link_t* link, pn_terminus_t* target, const std::string& name);
std::string generateName(pn_link_t*);
std::string qualifyName(const std::string&);
+ bool clear_pending(pn_delivery_t*);//tests and clears pending status for delivery
+ void abort_pending(pn_link_t*);//removes pending status for all deliveries associated with link
};
}}} // namespace qpid::broker::amqp
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org