From: rousskov <>
Date: Mon, 13 Aug 2007 22:48:20 +0000 (+0000)
Subject: Bug #2016 fix: Prevent BodyPipe async calls from getting seemingly
X-Git-Tag: SQUID_3_0_PRE7~84
X-Git-Url: http://git.ipfire.org/gitweb.cgi?a=commitdiff_plain;h=6c56baf69b075155b120cd442bd9290ce5f76bc7;p=thirdparty%2Fsquid.git

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 9293bed363..9997115b5c 100644
--- 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 96696fde15..0a336ff0b0 100644
--- 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