From 547140f366983b841429fd101440c05718e57e8d Mon Sep 17 00:00:00 2001
From: Amos Jeffries <amosjeffries@squid-cache.org>
Date: Thu, 23 Sep 2010 12:03:34 -0600
Subject: [PATCH] Author: Alex Rousskov <rousskov@measurement-factory.com>
 Added expectNoConsumption() BodyPipe method to allow the consumer side to
 tell the pipe that no more consumers are expected.

Calling expectNoConsumption() is needed when all consumers are gone but none
of them consumed anything. This is typical for HTTP server-side transactions
that abort before they receive the adapted ICAP request body. If FwdState
quits without calling expectNoConsumption(), the ICAP transaction supplying
the request body may get stuck waiting for buffer space. The body pipe may be
the ICAP only link with the master transaction once the header is handled.

This change is related to Squid bug #2964.
Based on lp 3p1-rock branch, r9609.
---
 src/BodyPipe.cc | 27 ++++++++++++++++++---------
 src/BodyPipe.h  |  2 ++
 2 files changed, 20 insertions(+), 9 deletions(-)

diff --git a/src/BodyPipe.cc b/src/BodyPipe.cc
index 799f81b499..ffc28c0803 100644
--- a/src/BodyPipe.cc
+++ b/src/BodyPipe.cc
@@ -129,7 +129,7 @@ void BodyConsumer::stopConsumingFrom(RefCount<BodyPipe> &pipe)
 BodyPipe::BodyPipe(Producer *aProducer): theBodySize(-1),
         theProducer(aProducer), theConsumer(0),
         thePutSize(0), theGetSize(0),
-        mustAutoConsume(false), isCheckedOut(false)
+        mustAutoConsume(false), abortedConsumption(false), isCheckedOut(false)
 {
     // TODO: teach MemBuf to start with zero minSize
     // TODO: limit maxSize by theBodySize, when known?
@@ -231,6 +231,8 @@ BodyPipe::setConsumerIfNotLate(const Consumer::Pointer &aConsumer)
         return false;
     }
 
+    Must(!abortedConsumption); // did not promise to never consume
+
     theConsumer = aConsumer;
     debugs(91,7, HERE << "set consumer" << status());
     if (theBuf.hasContent())
@@ -241,25 +243,30 @@ BodyPipe::setConsumerIfNotLate(const Consumer::Pointer &aConsumer)
     return true;
 }
 
-// When BodyPipe consumer is gone, all events for that consumer must not
-// reach the new consumer (if any). Otherwise, the calls may go out of order
-// (if _some_ calls are dropped due to the ultimate destination being
-// temporary NULL). The code keeps track of the number of outstanding
-// events and skips that number if consumer leaves. TODO: when AscyncCall
-// support is improved, should we just schedule calls directly to consumer?
 void
 BodyPipe::clearConsumer()
 {
     if (theConsumer.set()) {
         debugs(91,7, HERE << "clearing consumer" << status());
         theConsumer.clear();
-        if (consumedSize() && !exhausted()) {
+        // do not abort if we have not consumed so that HTTP or ICAP can retry
+        // benign xaction failures due to persistent connection race conditions
+        if (consumedSize())
+            expectNoConsumption();
+    }
+}
+
+void
+BodyPipe::expectNoConsumption()
+{
+    Must(!theConsumer);
+    if (!abortedConsumption && !exhausted()) {
             AsyncCall::Pointer call= asyncCall(91, 7,
                                                "BodyProducer::noteBodyConsumerAborted",
                                                BodyProducerDialer(theProducer,
                                                                   &BodyProducer::noteBodyConsumerAborted, this));
             ScheduleCallHere(call);
-        }
+            abortedConsumption = true;
     }
 }
 
@@ -433,6 +440,8 @@ const char *BodyPipe::status() const
 
     if (mustAutoConsume)
         outputBuffer.append(" A", 2);
+    if (abortedConsumption)
+        outputBuffer.append(" !C", 3);
     if (isCheckedOut)
         outputBuffer.append(" L", 2); // Locked
 
diff --git a/src/BodyPipe.h b/src/BodyPipe.h
index 47e30349d4..3cd2049545 100644
--- a/src/BodyPipe.h
+++ b/src/BodyPipe.h
@@ -109,6 +109,7 @@ public:
     // called by consumers
     bool setConsumerIfNotLate(const Consumer::Pointer &aConsumer);
     void clearConsumer(); // aborts if still piping
+    void expectNoConsumption(); ///< there will be no more setConsumer() calls
     size_t getMoreData(MemBuf &buf);
     void consume(size_t size);
     bool expectMoreAfter(uint64_t offset) const;
@@ -148,6 +149,7 @@ private:
     MemBuf theBuf; // produced but not yet consumed content, if any
 
     bool mustAutoConsume; // consume when there is no consumer
+    bool abortedConsumption; ///< called BodyProducer::noteBodyConsumerAborted
     bool isCheckedOut; // to keep track of checkout violations
 
     CBDATA_CLASS2(BodyPipe);
-- 
2.47.2