src/BodyPipe.cc

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