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
 193                 if (bodySize() != thePutSize)
 194                     debugs(91,3, HERE << "aborting on premature eof" << status());
 195         } else {
 196             // asserta that we can detect the abort if the consumer joins later
 197             assert(!bodySizeKnown() || bodySize() != thePutSize);
 198         }
 199         scheduleBodyEndNotification();
 200     }
 201 }
 202
 203 size_t
 204 BodyPipe::putMoreData(const char *buf, size_t size)
 205 {
 206     if (bodySizeKnown())
 207         size = min((uint64_t)size, unproducedSize());
 208
 209     const size_t spaceSize = static_cast<size_t>(theBuf.potentialSpaceSize());
 210     if ((size = min(size, spaceSize))) {
 211         theBuf.append(buf, size);
 212         postAppend(size);
 213         return size;
 214     }
 215     return 0;
 216 }
 217
 218 bool
 219 BodyPipe::setConsumerIfNotLate(Consumer *aConsumer)
 220 {
 221     assert(!theConsumer);
 222     assert(aConsumer);
 223
 224     // TODO: convert this into an exception and remove IfNotLate suffix
 225     // If there is something consumed already, we are in an auto-consuming mode
 226     // and it is too late to attach a real consumer to the pipe.
 227     if (theGetSize > 0) {
 228         assert(mustAutoConsume);
 229         return false;
 230     }
 231
 232     theConsumer = aConsumer;
 233     debugs(91,7, HERE << "set consumer" << status());
 234     if (theBuf.hasContent())
 235         scheduleBodyDataNotification();
 236     if (!theProducer)
 237         scheduleBodyEndNotification();
 238
 239     return true;
 240 }
 241
 242 // When BodyPipe consumer is gone, all events for that consumer must not
 243 // reach the new consumer (if any). Otherwise, the calls may go out of order
 244 // (if _some_ calls are dropped due to the ultimate destination being
 245 // temporary NULL). The code keeps track of the number of outstanding
 246 // events and skips that number if consumer leaves. TODO: when AscyncCall
 247 // support is improved, should we just schedule calls directly to consumer?
 248 void
 249 BodyPipe::clearConsumer()
 250 {
 251     if (theConsumer) {
 252         debugs(91,7, HERE << "clearing consumer" << status());
 253         theConsumer = NULL;
 254         if (consumedSize() && !exhausted()) {
 255             AsyncCall::Pointer call= asyncCall(91, 7,
 256                                                "BodyProducer::noteBodyConsumerAborted",
 257                                                BodyProducerDialer(theProducer,
 258                                                                   &BodyProducer::noteBodyConsumerAborted, this));
 259             ScheduleCallHere(call);
 260         }
 261     }
 262 }
 263
 264 size_t
 265 BodyPipe::getMoreData(MemBuf &buf)
 266 {
 267     if (!theBuf.hasContent())
 268         return 0; // did not touch the possibly uninitialized buf
 269
 270     if (buf.isNull())
 271         buf.init();
 272     const size_t size = min(theBuf.contentSize(), buf.potentialSpaceSize());
 273     buf.append(theBuf.content(), size);
 274     theBuf.consume(size);
 275     postConsume(size);
 276     return size; // cannot be zero if we called buf.init above
 277 }
 278
 279 void
 280 BodyPipe::consume(size_t size)
 281 {
 282     theBuf.consume(size);
 283     postConsume(size);
 284 }
 285
 286 // In the AutoConsumption  mode the consumer has gone but the producer continues
 287 // producing data. We are using a BodySink BodyConsumer which just discards the produced data.
 288 void
 289 BodyPipe::enableAutoConsumption()
 290 {
 291     mustAutoConsume = true;
 292     debugs(91,5, HERE << "enabled auto consumption" << status());
 293     if (!theConsumer && theBuf.hasContent())
 294         startAutoConsumption();
 295 }
 296
 297 // start auto consumption by creating body sink
 298 void
 299 BodyPipe::startAutoConsumption()
 300 {
 301     Must(mustAutoConsume);
 302     Must(!theConsumer);
 303     theConsumer = new BodySink;
 304     debugs(91,7, HERE << "starting auto consumption" << status());
 305     scheduleBodyDataNotification();
 306 }
 307
 308 MemBuf &
 309 BodyPipe::checkOut()
 310 {
 311     assert(!isCheckedOut);
 312     isCheckedOut = true;
 313     return theBuf;
 314 }
 315
 316 void
 317 BodyPipe::checkIn(Checkout &checkout)
 318 {
 319     assert(isCheckedOut);
 320     isCheckedOut = false;
 321     const size_t currentSize = theBuf.contentSize();
 322     if (checkout.checkedOutSize > currentSize)
 323         postConsume(checkout.checkedOutSize - currentSize);
 324     else
 325         if (checkout.checkedOutSize < currentSize)
 326             postAppend(currentSize - checkout.checkedOutSize);
 327 }
 328
 329 void
 330 BodyPipe::undoCheckOut(Checkout &checkout)
 331 {
 332     assert(isCheckedOut);
 333     const size_t currentSize = theBuf.contentSize();
 334     // We can only undo if size did not change, and even that carries
 335     // some risk. If this becomes a problem, the code checking out
 336     // raw buffers should always check them in (possibly unchanged)
 337     // instead of relying on the automated undo mechanism of Checkout.
 338     // The code can always use a temporary buffer to accomplish that.
 339     assert(checkout.checkedOutSize == currentSize);
 340 }
 341
 342 // TODO: Optimize: inform consumer/producer about more data/space only if
 343 // they used the data/space since we notified them last time.
 344
 345 void
 346 BodyPipe::postConsume(size_t size)
 347 {
 348     assert(!isCheckedOut);
 349     theGetSize += size;
 350     debugs(91,7, HERE << "consumed " << size << " bytes" << status());
 351     if (mayNeedMoreData()) {
 352         AsyncCall::Pointer call=  asyncCall(91, 7,
 353                                             "BodyProducer::noteMoreBodySpaceAvailable",
 354                                             BodyProducerDialer(theProducer,
 355                                                                &BodyProducer::noteMoreBodySpaceAvailable, this));
 356         ScheduleCallHere(call);
 357     }
 358 }
 359
 360 void
 361 BodyPipe::postAppend(size_t size)
 362 {
 363     assert(!isCheckedOut);
 364     thePutSize += size;
 365     debugs(91,7, HERE << "added " << size << " bytes" << status());
 366
 367     if (mustAutoConsume && !theConsumer && size > 0)
 368         startAutoConsumption();
 369
 370     // We should not consume here even if mustAutoConsume because the
 371     // caller may not be ready for the data to be consumed during this call.
 372     scheduleBodyDataNotification();
 373
 374     if (!mayNeedMoreData())
 375         clearProducer(true); // reached end-of-body
 376 }
 377
 378
 379 void
 380 BodyPipe::scheduleBodyDataNotification()
 381 {
 382     if (theConsumer) {
 383         AsyncCall::Pointer call = asyncCall(91, 7,
 384                                             "BodyConsumer::noteMoreBodyDataAvailable",
 385                                             BodyConsumerDialer(theConsumer,
 386                                                                &BodyConsumer::noteMoreBodyDataAvailable, this));
 387         ScheduleCallHere(call);
 388     }
 389 }
 390
 391 void
 392 BodyPipe::scheduleBodyEndNotification()
 393 {
 394     if (theConsumer) {
 395         if (bodySizeKnown() && bodySize() == thePutSize) {
 396             AsyncCall::Pointer call = asyncCall(91, 7,
 397                                                 "BodyConsumer::noteBodyProductionEnded",
 398                                                 BodyConsumerDialer(theConsumer,
 399                                                                    &BodyConsumer::noteBodyProductionEnded, this));
 400             ScheduleCallHere(call);
 401         } else {
 402             AsyncCall::Pointer call = asyncCall(91, 7,
 403                                                 "BodyConsumer::noteBodyProducerAborted",
 404                                                 BodyConsumerDialer(theConsumer,
 405                                                                    &BodyConsumer::noteBodyProducerAborted, this));
 406             ScheduleCallHere(call);
 407         }
 408     }
 409 }
 410
 411 // a short temporary string describing buffer status for debugging
 412 const char *BodyPipe::status() const
 413 {
 414     static MemBuf buf;
 415     buf.reset();
 416
 417     buf.append(" [", 2);
 418
 419     buf.Printf("%"PRIu64"<=%"PRIu64, theGetSize, thePutSize);
 420     if (theBodySize >= 0)
 421         buf.Printf("<=%"PRId64, theBodySize);
 422     else
 423         buf.append("<=?", 3);
 424
 425     buf.Printf(" %d+%d", (int)theBuf.contentSize(), (int)theBuf.spaceSize());
 426
 427     buf.Printf(" pipe%p", this);
 428     if (theProducer)
 429         buf.Printf(" prod%p", theProducer);
 430     if (theConsumer)
 431         buf.Printf(" cons%p", theConsumer);
 432
 433     if (mustAutoConsume)
 434         buf.append(" A", 2);
 435     if (isCheckedOut)
 436         buf.append(" L", 2); // Locked
 437
 438     buf.append("]", 1);
 439
 440     buf.terminate();
 441
 442     return buf.content();
 443 }
 444
 445
 446 /* BodyPipeCheckout */
 447
 448 BodyPipeCheckout::BodyPipeCheckout(BodyPipe &aPipe): pipe(aPipe),
 449         buf(aPipe.checkOut()), offset(aPipe.consumedSize()),
 450         checkedOutSize(buf.contentSize()), checkedIn(false)
 451 {
 452 }
 453
 454 BodyPipeCheckout::~BodyPipeCheckout()
 455 {
 456     if (!checkedIn)
 457         pipe.undoCheckOut(*this);
 458 }
 459
 460 void
 461 BodyPipeCheckout::checkIn()
 462 {
 463     assert(!checkedIn);
 464     pipe.checkIn(*this);
 465     checkedIn = true;
 466 }
 467
 468
 469 BodyPipeCheckout::BodyPipeCheckout(const BodyPipeCheckout &c): pipe(c.pipe),
 470         buf(c.buf), offset(c.offset), checkedOutSize(c.checkedOutSize),
 471         checkedIn(c.checkedIn)
 472 {
 473     assert(false); // prevent copying
 474 }
 475
 476 BodyPipeCheckout &
 477 BodyPipeCheckout::operator =(const BodyPipeCheckout &)
 478 {
 479     assert(false); // prevent assignment
 480     return *this;
 481 }