Bug #2016 fix: Prevent BodyPipe async calls from getting seemingly

out of order and causing truncated responses, especially with ICAP.

When BodyPipe consumer is gone, all async calls for that consumer must not
reach the next consumer (if any). Similarly, we should not schedule async calls
when no consumer has been registered yet.

Otherwise, the calls may go out of order if _some_ calls are dropped due to
the ultimate destination being temporary NULL. The new code does not schedule
async calls until consumer is registered. The new code also keeps track of the
number of outstanding events and skips that number if the consumer leaves.

TODO:  when AscyncCall support is improved, should we just schedule calls
directly to consumer? It could be a much cleaner solution than counting
pending calls and skipping them when needed.

diff --git a/src/BodyPipe.cc b/src/BodyPipe.cc
index 9293bed36388f3cc496cf7b67d5fcdc503d08d46..9997115b5c627609dd482b77b9702ea6372cd371 100644 (file)
--- a/src/BodyPipe.cc
+++ b/src/BodyPipe.cc
@@ -27,7 +27,8 @@ void BodyConsumer::stopConsumingFrom(RefCount<BodyPipe> &pipe)
 
 BodyPipe::BodyPipe(Producer *aProducer): theBodySize(-1),
        theProducer(aProducer), theConsumer(0),
-       thePutSize(0), theGetSize(0), mustAutoConsume(false), isCheckedOut(false)
+       thePutSize(0), theGetSize(0), theCCallsPending(0), theCCallsToSkip(0),
+       mustAutoConsume(false), isCheckedOut(false)
 {
        // TODO: teach MemBuf to start with zero minSize
        // TODO: limit maxSize by theBodySize, when known?
@@ -133,18 +134,25 @@ BodyPipe::setConsumerIfNotLate(Consumer *aConsumer)
        theConsumer = aConsumer;
        debugs(91,7, HERE << "set consumer" << status());
        if (theBuf.hasContent())
-               AsyncCall(91,5, this, BodyPipe::tellMoreBodyDataAvailable);
+               scheduleBodyDataNotification();
        if (!theProducer)
                scheduleBodyEndNotification();
 
        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) {
                debugs(91,7, HERE << "clearing consumer" << status());
                theConsumer = NULL;
+               theCCallsToSkip = theCCallsPending; // skip all pending consumer calls
                if (consumedSize() && !exhausted())
                        AsyncCall(91,5, this, BodyPipe::tellBodyConsumerAborted);
        }
@@ -177,7 +185,7 @@ BodyPipe::enableAutoConsumption() {
        mustAutoConsume = true;
        debugs(91,5, HERE << "enabled auto consumption" << status());
        if (!theConsumer && theBuf.hasContent())
-               AsyncCall(91,5, this, BodyPipe::tellMoreBodyDataAvailable);
+               scheduleBodyDataNotification();
 }
 
 MemBuf &
@@ -233,20 +241,32 @@ BodyPipe::postAppend(size_t size) {
 
        // We should not consume here even if mustAutoConsume because the
        // caller may not be ready for the data to be consumed during this call.
-       AsyncCall(91,5, this, BodyPipe::tellMoreBodyDataAvailable);
+       scheduleBodyDataNotification();
 
        if (!mayNeedMoreData())
                clearProducer(true); // reached end-of-body
 }
 
 
+void
+BodyPipe::scheduleBodyDataNotification()
+{
+       if (theConsumer || mustAutoConsume) {
+               ++theCCallsPending;
+               AsyncCall(91,5, this, BodyPipe::tellMoreBodyDataAvailable);
+       }
+}
+
 void
 BodyPipe::scheduleBodyEndNotification()
 {
-       if (bodySizeKnown() && bodySize() == thePutSize)
-               AsyncCall(91,5, this, BodyPipe::tellBodyProductionEnded);
-       else
-               AsyncCall(91,5, this, BodyPipe::tellBodyProducerAborted);
+       if (theConsumer) {
+               ++theCCallsPending;
+               if (bodySizeKnown() && bodySize() == thePutSize)
+                       AsyncCall(91,5, this, BodyPipe::tellBodyProductionEnded);
+               else
+                       AsyncCall(91,5, this, BodyPipe::tellBodyProducerAborted);
+       }
 }
 
 void BodyPipe::tellMoreBodySpaceAvailable()
@@ -263,6 +283,9 @@ void BodyPipe::tellBodyConsumerAborted()
 
 void BodyPipe::tellMoreBodyDataAvailable()
 {
+       if (skipCCall())
+               return;
+
        if (theConsumer != NULL)
                theConsumer->noteMoreBodyDataAvailable(*this);
        else
@@ -272,16 +295,36 @@ void BodyPipe::tellMoreBodyDataAvailable()
 
 void BodyPipe::tellBodyProductionEnded()
 {
+       if (skipCCall())
+               return;
+
        if (theConsumer != NULL)
                theConsumer->noteBodyProductionEnded(*this);
 }
 
 void BodyPipe::tellBodyProducerAborted()
 {
+       if (skipCCall())
+               return;
+
        if (theConsumer != NULL)
                theConsumer->noteBodyProducerAborted(*this);
 }
 
+// skips calls destined for the previous consumer; see BodyPipe::clearConsumer
+bool BodyPipe::skipCCall()
+{
+       assert(theCCallsPending > 0);
+       --theCCallsPending;
+
+       if (theCCallsToSkip <= 0)
+               return false;
+
+       --theCCallsToSkip;
+       debugs(91,5, HERE << "skipped call");
+       return true;
+}
+
 // a short temporary string describing buffer status for debugging
 const char *BodyPipe::status() const
 {

diff --git a/src/BodyPipe.h b/src/BodyPipe.h
index 96696fde15b3450f5947df2e1e9ad352f1908724..0a336ff0b02654d2380301e01038e913ffa6b822 100644 (file)
--- a/src/BodyPipe.h
+++ b/src/BodyPipe.h
@@ -115,12 +115,15 @@ class BodyPipe: public RefCountable {
                void checkIn(Checkout &checkout); // return updated raw buffer
                void undoCheckOut(Checkout &checkout); // undo checkout efffect
 
+               void scheduleBodyDataNotification();
                void scheduleBodyEndNotification();
 
                // keep counters in sync and notify producer or consumer
                void postConsume(size_t size);
                void postAppend(size_t size);
 
+               bool skipCCall(); // decides whether to skip the call, updates counters
+
        public: /* public to enable callbacks, but treat as private */
 
                /* these methods are calling producer and sibscriber note*()
@@ -147,6 +150,9 @@ class BodyPipe: public RefCountable {
                size_t thePutSize; // ever-increasing total
                size_t theGetSize; // ever-increasing total
 
+               int theCCallsPending; // outstanding calls to the consumer
+               int theCCallsToSkip; // how many calls to the consumer we should skip
+
                MemBuf theBuf; // produced but not yet consumed content, if any
 
                bool mustAutoConsume; // consume when there is no consumer