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.

diff --git a/src/BodyPipe.cc b/src/BodyPipe.cc
index 799f81b49952960cd790ea2405e04bfabba7ee1a..ffc28c0803a3df562f663def1f7a7930b67ad70f 100644 (file)
--- 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 47e30349d4d1400667b38fbdf4ae534a83847074..3cd204954531275a6901e4e6a7285431e53b6fa4 100644 (file)
--- 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);