src/BodyPipe.cc

   1
   2 #include "squid-old.h"
   3 #include "base/AsyncJobCalls.h"
   4 #include "base/TextException.h"
   5 #include "BodyPipe.h"
   6
   7 CBDATA_CLASS_INIT(BodyPipe);
   8
   9 // BodySink is a BodyConsumer class which  just consume and drops
  10 // data from a BodyPipe
  11 class BodySink: public BodyConsumer
  12 {
  13 public:
  14     BodySink(const BodyPipe::Pointer &bp): AsyncJob("BodySink"), body_pipe(bp) {}
  15     virtual ~BodySink() { assert(!body_pipe); }
  16
  17     virtual void noteMoreBodyDataAvailable(BodyPipe::Pointer bp) {
  18         size_t contentSize = bp->buf().contentSize();
  19         bp->consume(contentSize);
  20     }
  21     virtual void noteBodyProductionEnded(BodyPipe::Pointer bp) {
  22         stopConsumingFrom(body_pipe);
  23     }
  24     virtual void noteBodyProducerAborted(BodyPipe::Pointer bp) {
  25         stopConsumingFrom(body_pipe);
  26     }
  27     bool doneAll() const {return !body_pipe && AsyncJob::doneAll();}
  28
  29 private:
  30     BodyPipe::Pointer body_pipe; ///< the pipe we are consuming from
  31
  32     CBDATA_CLASS2(BodySink);
  33 };
  34
  35 CBDATA_CLASS_INIT(BodySink);
  36
  37 // The BodyProducerDialer is an AsyncCall class which used to schedule BodyProducer calls.
  38 // In addition to a normal AsyncCall checks if the BodyProducer is still the producer of
  39 // the BodyPipe passed as argument
  40 class BodyProducerDialer: public UnaryMemFunT<BodyProducer, BodyPipe::Pointer>
  41 {
  42 public:
  43     typedef UnaryMemFunT<BodyProducer, BodyPipe::Pointer> Parent;
  44
  45     BodyProducerDialer(const BodyProducer::Pointer &aProducer,
  46                        Parent::Method aHandler, BodyPipe::Pointer bp):
  47             Parent(aProducer, aHandler, bp) {}
  48
  49     virtual bool canDial(AsyncCall &call);
  50 };
  51
  52 // The BodyConsumerDialer is an AsyncCall class which used to schedule BodyConsumer calls.
  53 // In addition to a normal AsyncCall checks if the BodyConsumer is still the reciptient
  54 // of the BodyPipe passed as argument
  55 class BodyConsumerDialer: public UnaryMemFunT<BodyConsumer, BodyPipe::Pointer>
  56 {
  57 public:
  58     typedef UnaryMemFunT<BodyConsumer, BodyPipe::Pointer> Parent;
  59
  60     BodyConsumerDialer(const BodyConsumer::Pointer &aConsumer,
  61                        Parent::Method aHandler, BodyPipe::Pointer bp):
  62             Parent(aConsumer, aHandler, bp) {}
  63
  64     virtual bool canDial(AsyncCall &call);
  65 };
  66
  67 bool
  68 BodyProducerDialer::canDial(AsyncCall &call)
  69 {
  70     if (!Parent::canDial(call))
  71         return false;
  72
  73     const BodyProducer::Pointer &producer = job;
  74     BodyPipe::Pointer pipe = arg1;
  75     if (!pipe->stillProducing(producer)) {
  76         debugs(call.debugSection, call.debugLevel, HERE << producer <<
  77                " no longer producing for " << pipe->status());
  78         return call.cancel("no longer producing");
  79     }
  80
  81     return true;
  82 }
  83
  84 bool
  85 BodyConsumerDialer::canDial(AsyncCall &call)
  86 {
  87     if (!Parent::canDial(call))
  88         return false;
  89
  90     const BodyConsumer::Pointer &consumer = job;
  91     BodyPipe::Pointer pipe = arg1;
  92     if (!pipe->stillConsuming(consumer)) {
  93         debugs(call.debugSection, call.debugLevel, HERE << consumer <<
  94                " no longer consuming from " << pipe->status());
  95         return call.cancel("no longer consuming");
  96     }
  97
  98     return true;
  99 }
 100
 101
 102 /* BodyProducer */
 103
 104 // inform the pipe that we are done and clear the Pointer
 105 void BodyProducer::stopProducingFor(RefCount<BodyPipe> &pipe, bool atEof)
 106 {
 107     debugs(91,7, HERE << this << " will not produce for " << pipe <<
 108            "; atEof: " << atEof);
 109     assert(pipe != NULL); // be strict: the caller state may depend on this
 110     pipe->clearProducer(atEof);
 111     pipe = NULL;
 112 }
 113
 114
 115
 116 /* BodyConsumer */
 117
 118 // inform the pipe that we are done and clear the Pointer
 119 void BodyConsumer::stopConsumingFrom(RefCount<BodyPipe> &pipe)
 120 {
 121     debugs(91,7, HERE << this << " will not consume from " << pipe);
 122     assert(pipe != NULL); // be strict: the caller state may depend on this
 123     pipe->clearConsumer();
 124     pipe = NULL;
 125 }
 126
 127
 128 /* BodyPipe */
 129
 130 BodyPipe::BodyPipe(Producer *aProducer): theBodySize(-1),
 131         theProducer(aProducer), theConsumer(0),
 132         thePutSize(0), theGetSize(0),
 133         mustAutoConsume(false), abortedConsumption(false), isCheckedOut(false)
 134 {
 135     // TODO: teach MemBuf to start with zero minSize
 136     // TODO: limit maxSize by theBodySize, when known?
 137     theBuf.init(2*1024, MaxCapacity);
 138     debugs(91,7, HERE << "created BodyPipe" << status());
 139 }
 140
 141 BodyPipe::~BodyPipe()
 142 {
 143     debugs(91,7, HERE << "destroying BodyPipe" << status());
 144     assert(!theProducer);
 145     assert(!theConsumer);
 146     theBuf.clean();
 147 }
 148
 149 void BodyPipe::setBodySize(uint64_t aBodySize)
 150 {
 151     assert(!bodySizeKnown());
 152     assert(thePutSize <= aBodySize);
 153
 154     // If this assert fails, we need to add code to check for eof and inform
 155     // the consumer about the eof condition via scheduleBodyEndNotification,
 156     // because just setting a body size limit may trigger the eof condition.
 157     assert(!theConsumer);
 158
 159     theBodySize = aBodySize;
 160     debugs(91,7, HERE << "set body size" << status());
 161 }
 162
 163 uint64_t BodyPipe::bodySize() const
 164 {
 165     assert(bodySizeKnown());
 166     return static_cast<uint64_t>(theBodySize);
 167 }
 168
 169 bool BodyPipe::expectMoreAfter(uint64_t offset) const
 170 {
 171     assert(theGetSize <= offset);
 172     return offset < thePutSize || // buffer has more now or
 173            (!productionEnded() && mayNeedMoreData()); // buffer will have more
 174 }
 175
 176 bool BodyPipe::exhausted() const
 177 {
 178     return !expectMoreAfter(theGetSize);
 179 }
 180
 181 uint64_t BodyPipe::unproducedSize() const
 182 {
 183     return bodySize() - thePutSize; // bodySize() asserts that size is known
 184 }
 185
 186 void BodyPipe::expectProductionEndAfter(uint64_t size)
 187 {
 188     const uint64_t expectedSize = thePutSize + size;
 189     if (bodySizeKnown())
 190         Must(bodySize() == expectedSize);
 191     else
 192         theBodySize = expectedSize;
 193 }
 194
 195 void
 196 BodyPipe::clearProducer(bool atEof)
 197 {
 198     if (theProducer.set()) {
 199         debugs(91,7, HERE << "clearing BodyPipe producer" << status());
 200         theProducer.clear();
 201         if (atEof) {
 202             if (!bodySizeKnown())
 203                 theBodySize = thePutSize;
 204             else if (bodySize() != thePutSize)
 205                 debugs(91,3, HERE << "aborting on premature eof" << status());
 206         } else {
 207             // asserta that we can detect the abort if the consumer joins later
 208             assert(!bodySizeKnown() || bodySize() != thePutSize);
 209         }
 210         scheduleBodyEndNotification();
 211     }
 212 }
 213
 214 size_t
 215 BodyPipe::putMoreData(const char *aBuffer, size_t size)
 216 {
 217     if (bodySizeKnown())
 218         size = min((uint64_t)size, unproducedSize());
 219
 220     const size_t spaceSize = static_cast<size_t>(theBuf.potentialSpaceSize());
 221     if ((size = min(size, spaceSize))) {
 222         theBuf.append(aBuffer, size);
 223         postAppend(size);
 224         return size;
 225     }
 226     return 0;
 227 }
 228
 229 bool
 230 BodyPipe::setConsumerIfNotLate(const Consumer::Pointer &aConsumer)
 231 {
 232     assert(!theConsumer);
 233     assert(aConsumer.set()); // but might be invalid
 234
 235     // TODO: convert this into an exception and remove IfNotLate suffix
 236     // If there is something consumed already, we are in an auto-consuming mode
 237     // and it is too late to attach a real consumer to the pipe.
 238     if (theGetSize > 0) {
 239         assert(mustAutoConsume);
 240         return false;
 241     }
 242
 243     Must(!abortedConsumption); // did not promise to never consume
 244
 245     theConsumer = aConsumer;
 246     debugs(91,7, HERE << "set consumer" << status());
 247     if (theBuf.hasContent())
 248         scheduleBodyDataNotification();
 249     if (!theProducer)
 250         scheduleBodyEndNotification();
 251
 252     return true;
 253 }
 254
 255 void
 256 BodyPipe::clearConsumer()
 257 {
 258     if (theConsumer.set()) {
 259         debugs(91,7, HERE << "clearing consumer" << status());
 260         theConsumer.clear();
 261         // do not abort if we have not consumed so that HTTP or ICAP can retry
 262         // benign xaction failures due to persistent connection race conditions
 263         if (consumedSize())
 264             expectNoConsumption();
 265     }
 266 }
 267
 268 void
 269 BodyPipe::expectNoConsumption()
 270 {
 271     Must(!theConsumer);
 272     if (!abortedConsumption && !exhausted()) {
 273         AsyncCall::Pointer call= asyncCall(91, 7,
 274                                            "BodyProducer::noteBodyConsumerAborted",
 275                                            BodyProducerDialer(theProducer,
 276                                                               &BodyProducer::noteBodyConsumerAborted, this));
 277         ScheduleCallHere(call);
 278         abortedConsumption = true;
 279     }
 280 }
 281
 282 size_t
 283 BodyPipe::getMoreData(MemBuf &aMemBuffer)
 284 {
 285     if (!theBuf.hasContent())
 286         return 0; // did not touch the possibly uninitialized buf
 287
 288     if (aMemBuffer.isNull())
 289         aMemBuffer.init();
 290     const size_t size = min(theBuf.contentSize(), aMemBuffer.potentialSpaceSize());
 291     aMemBuffer.append(theBuf.content(), size);
 292     theBuf.consume(size);
 293     postConsume(size);
 294     return size; // cannot be zero if we called buf.init above
 295 }
 296
 297 void
 298 BodyPipe::consume(size_t size)
 299 {
 300     theBuf.consume(size);
 301     postConsume(size);
 302 }
 303
 304 // In the AutoConsumption  mode the consumer has gone but the producer continues
 305 // producing data. We are using a BodySink BodyConsumer which just discards the produced data.
 306 void
 307 BodyPipe::enableAutoConsumption()
 308 {
 309     mustAutoConsume = true;
 310     debugs(91,5, HERE << "enabled auto consumption" << status());
 311     if (!theConsumer && theBuf.hasContent())
 312         startAutoConsumption();
 313 }
 314
 315 // start auto consumption by creating body sink
 316 void
 317 BodyPipe::startAutoConsumption()
 318 {
 319     Must(mustAutoConsume);
 320     Must(!theConsumer);
 321     theConsumer = new BodySink(this);
 322     debugs(91,7, HERE << "starting auto consumption" << status());
 323     scheduleBodyDataNotification();
 324 }
 325
 326 MemBuf &
 327 BodyPipe::checkOut()
 328 {
 329     assert(!isCheckedOut);
 330     isCheckedOut = true;
 331     return theBuf;
 332 }
 333
 334 void
 335 BodyPipe::checkIn(Checkout &checkout)
 336 {
 337     assert(isCheckedOut);
 338     isCheckedOut = false;
 339     const size_t currentSize = theBuf.contentSize();
 340     if (checkout.checkedOutSize > currentSize)
 341         postConsume(checkout.checkedOutSize - currentSize);
 342     else if (checkout.checkedOutSize < currentSize)
 343         postAppend(currentSize - checkout.checkedOutSize);
 344 }
 345
 346 void
 347 BodyPipe::undoCheckOut(Checkout &checkout)
 348 {
 349     assert(isCheckedOut);
 350     const size_t currentSize = theBuf.contentSize();
 351     // We can only undo if size did not change, and even that carries
 352     // some risk. If this becomes a problem, the code checking out
 353     // raw buffers should always check them in (possibly unchanged)
 354     // instead of relying on the automated undo mechanism of Checkout.
 355     // The code can always use a temporary buffer to accomplish that.
 356     Must(checkout.checkedOutSize == currentSize);
 357 }
 358
 359 // TODO: Optimize: inform consumer/producer about more data/space only if
 360 // they used the data/space since we notified them last time.
 361
 362 void
 363 BodyPipe::postConsume(size_t size)
 364 {
 365     assert(!isCheckedOut);
 366     theGetSize += size;
 367     debugs(91,7, HERE << "consumed " << size << " bytes" << status());
 368     if (mayNeedMoreData()) {
 369         AsyncCall::Pointer call=  asyncCall(91, 7,
 370                                             "BodyProducer::noteMoreBodySpaceAvailable",
 371                                             BodyProducerDialer(theProducer,
 372                                                                &BodyProducer::noteMoreBodySpaceAvailable, this));
 373         ScheduleCallHere(call);
 374     }
 375 }
 376
 377 void
 378 BodyPipe::postAppend(size_t size)
 379 {
 380     assert(!isCheckedOut);
 381     thePutSize += size;
 382     debugs(91,7, HERE << "added " << size << " bytes" << status());
 383
 384     if (mustAutoConsume && !theConsumer && size > 0)
 385         startAutoConsumption();
 386
 387     // We should not consume here even if mustAutoConsume because the
 388     // caller may not be ready for the data to be consumed during this call.
 389     scheduleBodyDataNotification();
 390
 391     if (!mayNeedMoreData())
 392         clearProducer(true); // reached end-of-body
 393 }
 394
 395
 396 void
 397 BodyPipe::scheduleBodyDataNotification()
 398 {
 399     if (theConsumer.valid()) { // TODO: allow asyncCall() to check this instead
 400         AsyncCall::Pointer call = asyncCall(91, 7,
 401                                             "BodyConsumer::noteMoreBodyDataAvailable",
 402                                             BodyConsumerDialer(theConsumer,
 403                                                                &BodyConsumer::noteMoreBodyDataAvailable, this));
 404         ScheduleCallHere(call);
 405     }
 406 }
 407
 408 void
 409 BodyPipe::scheduleBodyEndNotification()
 410 {
 411     if (theConsumer.valid()) { // TODO: allow asyncCall() to check this instead
 412         if (bodySizeKnown() && bodySize() == thePutSize) {
 413             AsyncCall::Pointer call = asyncCall(91, 7,
 414                                                 "BodyConsumer::noteBodyProductionEnded",
 415                                                 BodyConsumerDialer(theConsumer,
 416                                                                    &BodyConsumer::noteBodyProductionEnded, this));
 417             ScheduleCallHere(call);
 418         } else {
 419             AsyncCall::Pointer call = asyncCall(91, 7,
 420                                                 "BodyConsumer::noteBodyProducerAborted",
 421                                                 BodyConsumerDialer(theConsumer,
 422                                                                    &BodyConsumer::noteBodyProducerAborted, this));
 423             ScheduleCallHere(call);
 424         }
 425     }
 426 }
 427
 428 // a short temporary string describing buffer status for debugging
 429 const char *BodyPipe::status() const
 430 {
 431     static MemBuf outputBuffer;
 432     outputBuffer.reset();
 433
 434     outputBuffer.append(" [", 2);
 435
 436     outputBuffer.Printf("%"PRIu64"<=%"PRIu64, theGetSize, thePutSize);
 437     if (theBodySize >= 0)
 438         outputBuffer.Printf("<=%"PRId64, theBodySize);
 439     else
 440         outputBuffer.append("<=?", 3);
 441
 442     outputBuffer.Printf(" %d+%d", (int)theBuf.contentSize(), (int)theBuf.spaceSize());
 443
 444     outputBuffer.Printf(" pipe%p", this);
 445     if (theProducer.set())
 446         outputBuffer.Printf(" prod%p", theProducer.get());
 447     if (theConsumer.set())
 448         outputBuffer.Printf(" cons%p", theConsumer.get());
 449
 450     if (mustAutoConsume)
 451         outputBuffer.append(" A", 2);
 452     if (abortedConsumption)
 453         outputBuffer.append(" !C", 3);
 454     if (isCheckedOut)
 455         outputBuffer.append(" L", 2); // Locked
 456
 457     outputBuffer.append("]", 1);
 458
 459     outputBuffer.terminate();
 460
 461     return outputBuffer.content();
 462 }
 463
 464
 465 /* BodyPipeCheckout */
 466
 467 BodyPipeCheckout::BodyPipeCheckout(BodyPipe &aPipe): pipe(aPipe),
 468         buf(aPipe.checkOut()), offset(aPipe.consumedSize()),
 469         checkedOutSize(buf.contentSize()), checkedIn(false)
 470 {
 471 }
 472
 473 BodyPipeCheckout::~BodyPipeCheckout()
 474 {
 475     if (!checkedIn) {
 476         // Do not pipe.undoCheckOut(*this) because it asserts or throws
 477         // TODO: consider implementing the long-term solution discussed at
 478         // http://www.mail-archive.com/squid-dev@squid-cache.org/msg07910.html
 479         debugs(91,2, HERE << "Warning: cannot undo BodyPipeCheckout");
 480         pipe.checkIn(*this);
 481     }
 482 }
 483
 484 void
 485 BodyPipeCheckout::checkIn()
 486 {
 487     assert(!checkedIn);
 488     pipe.checkIn(*this);
 489     checkedIn = true;
 490 }
 491
 492
 493 BodyPipeCheckout::BodyPipeCheckout(const BodyPipeCheckout &c): pipe(c.pipe),
 494         buf(c.buf), offset(c.offset), checkedOutSize(c.checkedOutSize),
 495         checkedIn(c.checkedIn)
 496 {
 497     assert(false); // prevent copying
 498 }
 499
 500 BodyPipeCheckout &
 501 BodyPipeCheckout::operator =(const BodyPipeCheckout &)
 502 {
 503     assert(false); // prevent assignment
 504     return *this;
 505 }