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

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1710683 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 48 insertions, 6 deletions

diff --git a/qpid/cpp/src/qpid/broker/amqp/Incoming.cpp b/qpid/cpp/src/qpid/broker/amqp/Incoming.cpp
index 3986818846..fd3319eca1 100644
--- a/qpid/cpp/src/qpid/broker/amqp/Incoming.cpp
+++ b/qpid/cpp/src/qpid/broker/amqp/Incoming.cpp
@@ -150,6 +150,7 @@ void DecodingIncoming::deliver(boost::intrusive_ptr<qpid::broker::amqp::Message>
     received->begin();
     handle(message, session.getTransaction(delivery));
     Transfer t(delivery, sessionPtr);
+    sessionPtr->pending_accept(delivery);
     received->end(t);
 }
 }}} // namespace qpid::broker::amqp
diff --git a/qpid/cpp/src/qpid/broker/amqp/Session.cpp b/qpid/cpp/src/qpid/broker/amqp/Session.cpp
index aa4ba03dfd..7780456bd1 100644
--- a/qpid/cpp/src/qpid/broker/amqp/Session.cpp
+++ b/qpid/cpp/src/qpid/broker/amqp/Session.cpp
@@ -607,6 +607,7 @@ void Session::detach(pn_link_t* link, bool closed)
     } 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, bool closed)
     }
 }
 
+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(boost::intrusive_ptr<qpid::broker::amqp::Mes
                 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);
                 }
 
diff --git a/qpid/cpp/src/qpid/broker/amqp/Session.h b/qpid/cpp/src/qpid/broker/amqp/Session.h
index 2537d6ca27..83376064c7 100644
--- a/qpid/cpp/src/qpid/broker/amqp/Session.h
+++ b/qpid/cpp/src/qpid/broker/amqp/Session.h
@@ -77,8 +77,10 @@ class Session : public ManagedSession, public boost::enable_shared_from_this<Ses
      */
     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, public boost::enable_shared_from_this<Ses
     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, public boost::enable_shared_from_this<Ses
     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