src/Server.cc

   1 /*
   2  * DEBUG:
   3  * AUTHOR: Duane Wessels
   4  *
   5  * SQUID Web Proxy Cache          http://www.squid-cache.org/
   6  * ----------------------------------------------------------
   7  *
   8  *  Squid is the result of efforts by numerous individuals from
   9  *  the Internet community; see the CONTRIBUTORS file for full
  10  *  details.   Many organizations have provided support for Squid's
  11  *  development; see the SPONSORS file for full details.  Squid is
  12  *  Copyrighted (C) 2001 by the Regents of the University of
  13  *  California; see the COPYRIGHT file for full details.  Squid
  14  *  incorporates software developed and/or copyrighted by other
  15  *  sources; see the CREDITS file for full details.
  16  *
  17  *  This program is free software; you can redistribute it and/or modify
  18  *  it under the terms of the GNU General Public License as published by
  19  *  the Free Software Foundation; either version 2 of the License, or
  20  *  (at your option) any later version.
  21  *
  22  *  This program is distributed in the hope that it will be useful,
  23  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  24  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  25  *  GNU General Public License for more details.
  26  *
  27  *  You should have received a copy of the GNU General Public License
  28  *  along with this program; if not, write to the Free Software
  29  *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111, USA.
  30  *
  31  */
  32
  33 #include "squid.h"
  34 #include "acl/FilledChecklist.h"
  35 #include "acl/Gadgets.h"
  36 #include "base/TextException.h"
  37 #include "comm/Connection.h"
  38 #include "comm/forward.h"
  39 #include "comm/Write.h"
  40 #include "err_detail_type.h"
  41 #include "errorpage.h"
  42 #include "fd.h"
  43 #include "HttpHdrContRange.h"
  44 #include "HttpReply.h"
  45 #include "HttpRequest.h"
  46 #include "Server.h"
  47 #include "SquidConfig.h"
  48 #include "SquidTime.h"
  49 #include "StatCounters.h"
  50 #include "Store.h"
  51 #include "tools.h"
  52 #include "URL.h"
  53
  54 #if USE_ADAPTATION
  55 #include "adaptation/AccessCheck.h"
  56 #include "adaptation/Answer.h"
  57 #include "adaptation/Iterator.h"
  58 #include "base/AsyncCall.h"
  59 #endif
  60
  61 // implemented in client_side_reply.cc until sides have a common parent
  62 void purgeEntriesByUrl(HttpRequest * req, const char *url);
  63
  64 ServerStateData::ServerStateData(FwdState *theFwdState): AsyncJob("ServerStateData"),
  65         requestSender(NULL),
  66 #if USE_ADAPTATION
  67         adaptedHeadSource(NULL),
  68         adaptationAccessCheckPending(false),
  69         startedAdaptation(false),
  70 #endif
  71         receivedWholeRequestBody(false),
  72         theVirginReply(NULL),
  73         theFinalReply(NULL)
  74 {
  75     fwd = theFwdState;
  76     entry = fwd->entry;
  77
  78     entry->lock("ServerStateData");
  79
  80     request = fwd->request;
  81     HTTPMSGLOCK(request);
  82 }
  83
  84 ServerStateData::~ServerStateData()
  85 {
  86     // paranoid: check that swanSong has been called
  87     assert(!requestBodySource);
  88 #if USE_ADAPTATION
  89     assert(!virginBodyDestination);
  90     assert(!adaptedBodySource);
  91 #endif
  92
  93     entry->unlock("ServerStateData");
  94
  95     HTTPMSGUNLOCK(request);
  96     HTTPMSGUNLOCK(theVirginReply);
  97     HTTPMSGUNLOCK(theFinalReply);
  98
  99     fwd = NULL; // refcounted
 100
 101     if (responseBodyBuffer != NULL) {
 102         delete responseBodyBuffer;
 103         responseBodyBuffer = NULL;
 104     }
 105 }
 106
 107 void
 108 ServerStateData::swanSong()
 109 {
 110     // get rid of our piping obligations
 111     if (requestBodySource != NULL)
 112         stopConsumingFrom(requestBodySource);
 113
 114 #if USE_ADAPTATION
 115     cleanAdaptation();
 116 #endif
 117
 118     BodyConsumer::swanSong();
 119 #if USE_ADAPTATION
 120     Initiator::swanSong();
 121     BodyProducer::swanSong();
 122 #endif
 123
 124     // paranoid: check that swanSong has been called
 125     // extra paranoid: yeah, I really mean it. they MUST pass here.
 126     assert(!requestBodySource);
 127 #if USE_ADAPTATION
 128     assert(!virginBodyDestination);
 129     assert(!adaptedBodySource);
 130 #endif
 131 }
 132
 133 HttpReply *
 134 ServerStateData::virginReply()
 135 {
 136     assert(theVirginReply);
 137     return theVirginReply;
 138 }
 139
 140 const HttpReply *
 141 ServerStateData::virginReply() const
 142 {
 143     assert(theVirginReply);
 144     return theVirginReply;
 145 }
 146
 147 HttpReply *
 148 ServerStateData::setVirginReply(HttpReply *rep)
 149 {
 150     debugs(11,5, HERE << this << " setting virgin reply to " << rep);
 151     assert(!theVirginReply);
 152     assert(rep);
 153     theVirginReply = rep;
 154     HTTPMSGLOCK(theVirginReply);
 155     return theVirginReply;
 156 }
 157
 158 HttpReply *
 159 ServerStateData::finalReply()
 160 {
 161     assert(theFinalReply);
 162     return theFinalReply;
 163 }
 164
 165 HttpReply *
 166 ServerStateData::setFinalReply(HttpReply *rep)
 167 {
 168     debugs(11,5, HERE << this << " setting final reply to " << rep);
 169
 170     assert(!theFinalReply);
 171     assert(rep);
 172     theFinalReply = rep;
 173     HTTPMSGLOCK(theFinalReply);
 174
 175     // give entry the reply because haveParsedReplyHeaders() expects it there
 176     entry->replaceHttpReply(theFinalReply, false); // but do not write yet
 177     haveParsedReplyHeaders(); // update the entry/reply (e.g., set timestamps)
 178     if (!EBIT_TEST(entry->flags, RELEASE_REQUEST) && blockCaching())
 179         entry->release();
 180     entry->startWriting(); // write the updated entry to store
 181
 182     return theFinalReply;
 183 }
 184
 185 // called when no more server communication is expected; may quit
 186 void
 187 ServerStateData::serverComplete()
 188 {
 189     debugs(11,5,HERE << "serverComplete " << this);
 190
 191     if (!doneWithServer()) {
 192         closeServer();
 193         assert(doneWithServer());
 194     }
 195
 196     completed = true;
 197
 198     HttpRequest *r = originalRequest();
 199     r->hier.total_response_time = r->hier.first_conn_start.tv_sec ?
 200                                   tvSubMsec(r->hier.first_conn_start, current_time) : -1;
 201
 202     if (requestBodySource != NULL)
 203         stopConsumingFrom(requestBodySource);
 204
 205     if (responseBodyBuffer != NULL)
 206         return;
 207
 208     serverComplete2();
 209 }
 210
 211 void
 212 ServerStateData::serverComplete2()
 213 {
 214     debugs(11,5,HERE << "serverComplete2 " << this);
 215
 216 #if USE_ADAPTATION
 217     if (virginBodyDestination != NULL)
 218         stopProducingFor(virginBodyDestination, true);
 219
 220     if (!doneWithAdaptation())
 221         return;
 222 #endif
 223
 224     completeForwarding();
 225 }
 226
 227 bool ServerStateData::doneAll() const
 228 {
 229     return  doneWithServer() &&
 230 #if USE_ADAPTATION
 231             doneWithAdaptation() &&
 232             Adaptation::Initiator::doneAll() &&
 233             BodyProducer::doneAll() &&
 234 #endif
 235             BodyConsumer::doneAll();
 236 }
 237
 238 // FTP side overloads this to work around multiple calls to fwd->complete
 239 void
 240 ServerStateData::completeForwarding()
 241 {
 242     debugs(11,5, HERE << "completing forwarding for "  << fwd);
 243     assert(fwd != NULL);
 244     fwd->complete();
 245 }
 246
 247 // Register to receive request body
 248 bool ServerStateData::startRequestBodyFlow()
 249 {
 250     HttpRequest *r = originalRequest();
 251     assert(r->body_pipe != NULL);
 252     requestBodySource = r->body_pipe;
 253     if (requestBodySource->setConsumerIfNotLate(this)) {
 254         debugs(11,3, HERE << "expecting request body from " <<
 255                requestBodySource->status());
 256         return true;
 257     }
 258
 259     debugs(11,3, HERE << "aborting on partially consumed request body: " <<
 260            requestBodySource->status());
 261     requestBodySource = NULL;
 262     return false;
 263 }
 264
 265 // Entry-dependent callbacks use this check to quit if the entry went bad
 266 bool
 267 ServerStateData::abortOnBadEntry(const char *abortReason)
 268 {
 269     if (entry->isAccepting())
 270         return false;
 271
 272     debugs(11,5, HERE << "entry is not Accepting!");
 273     abortTransaction(abortReason);
 274     return true;
 275 }
 276
 277 // more request or adapted response body is available
 278 void
 279 ServerStateData::noteMoreBodyDataAvailable(BodyPipe::Pointer bp)
 280 {
 281 #if USE_ADAPTATION
 282     if (adaptedBodySource == bp) {
 283         handleMoreAdaptedBodyAvailable();
 284         return;
 285     }
 286 #endif
 287     if (requestBodySource == bp)
 288         handleMoreRequestBodyAvailable();
 289 }
 290
 291 // the entire request or adapted response body was provided, successfully
 292 void
 293 ServerStateData::noteBodyProductionEnded(BodyPipe::Pointer bp)
 294 {
 295 #if USE_ADAPTATION
 296     if (adaptedBodySource == bp) {
 297         handleAdaptedBodyProductionEnded();
 298         return;
 299     }
 300 #endif
 301     if (requestBodySource == bp)
 302         handleRequestBodyProductionEnded();
 303 }
 304
 305 // premature end of the request or adapted response body production
 306 void
 307 ServerStateData::noteBodyProducerAborted(BodyPipe::Pointer bp)
 308 {
 309 #if USE_ADAPTATION
 310     if (adaptedBodySource == bp) {
 311         handleAdaptedBodyProducerAborted();
 312         return;
 313     }
 314 #endif
 315     if (requestBodySource == bp)
 316         handleRequestBodyProducerAborted();
 317 }
 318
 319 // more origin request body data is available
 320 void
 321 ServerStateData::handleMoreRequestBodyAvailable()
 322 {
 323     if (!requestSender)
 324         sendMoreRequestBody();
 325     else
 326         debugs(9,3, HERE << "waiting for request body write to complete");
 327 }
 328
 329 // there will be no more handleMoreRequestBodyAvailable calls
 330 void
 331 ServerStateData::handleRequestBodyProductionEnded()
 332 {
 333     receivedWholeRequestBody = true;
 334     if (!requestSender)
 335         doneSendingRequestBody();
 336     else
 337         debugs(9,3, HERE << "waiting for request body write to complete");
 338 }
 339
 340 // called when we are done sending request body; kids extend this
 341 void
 342 ServerStateData::doneSendingRequestBody()
 343 {
 344     debugs(9,3, HERE << "done sending request body");
 345     assert(requestBodySource != NULL);
 346     stopConsumingFrom(requestBodySource);
 347
 348     // kids extend this
 349 }
 350
 351 // called when body producers aborts; kids extend this
 352 void
 353 ServerStateData::handleRequestBodyProducerAborted()
 354 {
 355     if (requestSender != NULL)
 356         debugs(9,3, HERE << "fyi: request body aborted while we were sending");
 357
 358     fwd->dontRetry(true); // the problem is not with the server
 359     stopConsumingFrom(requestBodySource); // requestSender, if any, will notice
 360
 361     // kids extend this
 362 }
 363
 364 // called when we wrote request headers(!) or a part of the body
 365 void
 366 ServerStateData::sentRequestBody(const CommIoCbParams &io)
 367 {
 368     debugs(11, 5, "sentRequestBody: FD " << io.fd << ": size " << io.size << ": errflag " << io.flag << ".");
 369     debugs(32,3,HERE << "sentRequestBody called");
 370
 371     requestSender = NULL;
 372
 373     if (io.size > 0) {
 374         fd_bytes(io.fd, io.size, FD_WRITE);
 375         kb_incr(&(statCounter.server.all.kbytes_out), io.size);
 376         // kids should increment their counters
 377     }
 378
 379     if (io.flag == COMM_ERR_CLOSING)
 380         return;
 381
 382     if (!requestBodySource) {
 383         debugs(9,3, HERE << "detected while-we-were-sending abort");
 384         return; // do nothing;
 385     }
 386
 387     if (io.flag) {
 388         debugs(11, DBG_IMPORTANT, "sentRequestBody error: FD " << io.fd << ": " << xstrerr(io.xerrno));
 389         ErrorState *err;
 390         err = new ErrorState(ERR_WRITE_ERROR, Http::scBadGateway, fwd->request);
 391         err->xerrno = io.xerrno;
 392         fwd->fail(err);
 393         abortTransaction("I/O error while sending request body");
 394         return;
 395     }
 396
 397     if (EBIT_TEST(entry->flags, ENTRY_ABORTED)) {
 398         abortTransaction("store entry aborted while sending request body");
 399         return;
 400     }
 401
 402     if (!requestBodySource->exhausted())
 403         sendMoreRequestBody();
 404     else if (receivedWholeRequestBody)
 405         doneSendingRequestBody();
 406     else
 407         debugs(9,3, HERE << "waiting for body production end or abort");
 408 }
 409
 410 void
 411 ServerStateData::sendMoreRequestBody()
 412 {
 413     assert(requestBodySource != NULL);
 414     assert(!requestSender);
 415
 416     const Comm::ConnectionPointer conn = dataConnection();
 417
 418     if (!Comm::IsConnOpen(conn)) {
 419         debugs(9,3, HERE << "cannot send request body to closing " << conn);
 420         return; // wait for the kid's close handler; TODO: assert(closer);
 421     }
 422
 423     MemBuf buf;
 424     if (getMoreRequestBody(buf) && buf.contentSize() > 0) {
 425         debugs(9,3, HERE << "will write " << buf.contentSize() << " request body bytes");
 426         typedef CommCbMemFunT<ServerStateData, CommIoCbParams> Dialer;
 427         requestSender = JobCallback(93,3, Dialer, this, ServerStateData::sentRequestBody);
 428         Comm::Write(conn, &buf, requestSender);
 429     } else {
 430         debugs(9,3, HERE << "will wait for more request body bytes or eof");
 431         requestSender = NULL;
 432     }
 433 }
 434
 435 /// either fill buf with available [encoded] request body bytes or return false
 436 bool
 437 ServerStateData::getMoreRequestBody(MemBuf &buf)
 438 {
 439     // default implementation does not encode request body content
 440     Must(requestBodySource != NULL);
 441     return requestBodySource->getMoreData(buf);
 442 }
 443
 444 // Compares hosts in urls, returns false if different, no sheme, or no host.
 445 static bool
 446 sameUrlHosts(const char *url1, const char *url2)
 447 {
 448     // XXX: Want urlHostname() here, but it uses static storage and copying
 449     const char *host1 = strchr(url1, ':');
 450     const char *host2 = strchr(url2, ':');
 451
 452     if (host1 && host2) {
 453         // skip scheme slashes
 454         do {
 455             ++host1;
 456             ++host2;
 457         } while (*host1 == '/' && *host2 == '/');
 458
 459         if (!*host1)
 460             return false; // no host
 461
 462         // increment while the same until we reach the end of the URL/host
 463         while (*host1 && *host1 != '/' && *host1 == *host2) {
 464             ++host1;
 465             ++host2;
 466         }
 467         return *host1 == *host2;
 468     }
 469
 470     return false; // no URL scheme
 471 }
 472
 473 // purges entries that match the value of a given HTTP [response] header
 474 static void
 475 purgeEntriesByHeader(HttpRequest *req, const char *reqUrl, HttpMsg *rep, http_hdr_type hdr)
 476 {
 477     const char *hdrUrl, *absUrl;
 478
 479     absUrl = NULL;
 480     hdrUrl = rep->header.getStr(hdr);
 481     if (hdrUrl == NULL) {
 482         return;
 483     }
 484
 485     /*
 486      * If the URL is relative, make it absolute so we can find it.
 487      * If it's absolute, make sure the host parts match to avoid DOS attacks
 488      * as per RFC 2616 13.10.
 489      */
 490     if (urlIsRelative(hdrUrl)) {
 491         absUrl = urlMakeAbsolute(req, hdrUrl);
 492         if (absUrl != NULL) {
 493             hdrUrl = absUrl;
 494         }
 495     } else if (!sameUrlHosts(reqUrl, hdrUrl)) {
 496         return;
 497     }
 498
 499     purgeEntriesByUrl(req, hdrUrl);
 500
 501     if (absUrl != NULL) {
 502         safe_free(absUrl);
 503     }
 504 }
 505
 506 // some HTTP methods should purge matching cache entries
 507 void
 508 ServerStateData::maybePurgeOthers()
 509 {
 510     // only some HTTP methods should purge matching cache entries
 511     if (!request->method.purgesOthers())
 512         return;
 513
 514     // and probably only if the response was successful
 515     if (theFinalReply->sline.status() >= 400)
 516         return;
 517
 518     // XXX: should we use originalRequest() here?
 519     const char *reqUrl = urlCanonical(request);
 520     debugs(88, 5, "maybe purging due to " << RequestMethodStr(request->method) << ' ' << reqUrl);
 521     purgeEntriesByUrl(request, reqUrl);
 522     purgeEntriesByHeader(request, reqUrl, theFinalReply, HDR_LOCATION);
 523     purgeEntriesByHeader(request, reqUrl, theFinalReply, HDR_CONTENT_LOCATION);
 524 }
 525
 526 /// called when we have final (possibly adapted) reply headers; kids extend
 527 void
 528 ServerStateData::haveParsedReplyHeaders()
 529 {
 530     Must(theFinalReply);
 531     maybePurgeOthers();
 532
 533     // adaptation may overwrite old offset computed using the virgin response
 534     const bool partial = theFinalReply->content_range &&
 535                          theFinalReply->sline.status() == Http::scPartialContent;
 536     currentOffset = partial ? theFinalReply->content_range->spec.offset : 0;
 537 }
 538
 539 /// whether to prevent caching of an otherwise cachable response
 540 bool
 541 ServerStateData::blockCaching()
 542 {
 543     if (const Acl::Tree *acl = Config.accessList.storeMiss) {
 544         // This relatively expensive check is not in StoreEntry::checkCachable:
 545         // That method lacks HttpRequest and may be called too many times.
 546         ACLFilledChecklist ch(acl, originalRequest(), NULL);
 547         ch.reply = const_cast<HttpReply*>(entry->getReply()); // ACLFilledChecklist API bug
 548         HTTPMSGLOCK(ch.reply);
 549         if (ch.fastCheck() != ACCESS_ALLOWED) { // when in doubt, block
 550             debugs(20, 3, "store_miss prohibits caching");
 551             return true;
 552         }
 553     }
 554     return false;
 555 }
 556
 557 HttpRequest *
 558 ServerStateData::originalRequest()
 559 {
 560     return request;
 561 }
 562
 563 #if USE_ADAPTATION
 564 /// Initiate an asynchronous adaptation transaction which will call us back.
 565 void
 566 ServerStateData::startAdaptation(const Adaptation::ServiceGroupPointer &group, HttpRequest *cause)
 567 {
 568     debugs(11, 5, "ServerStateData::startAdaptation() called");
 569     // check whether we should be sending a body as well
 570     // start body pipe to feed ICAP transaction if needed
 571     assert(!virginBodyDestination);
 572     HttpReply *vrep = virginReply();
 573     assert(!vrep->body_pipe);
 574     int64_t size = 0;
 575     if (vrep->expectingBody(cause->method, size) && size) {
 576         virginBodyDestination = new BodyPipe(this);
 577         vrep->body_pipe = virginBodyDestination;
 578         debugs(93, 6, HERE << "will send virgin reply body to " <<
 579                virginBodyDestination << "; size: " << size);
 580         if (size > 0)
 581             virginBodyDestination->setBodySize(size);
 582     }
 583
 584     adaptedHeadSource = initiateAdaptation(
 585                             new Adaptation::Iterator(vrep, cause, fwd->al, group));
 586     startedAdaptation = initiated(adaptedHeadSource);
 587     Must(startedAdaptation);
 588 }
 589
 590 // properly cleans up ICAP-related state
 591 // may be called multiple times
 592 void ServerStateData::cleanAdaptation()
 593 {
 594     debugs(11,5, HERE << "cleaning ICAP; ACL: " << adaptationAccessCheckPending);
 595
 596     if (virginBodyDestination != NULL)
 597         stopProducingFor(virginBodyDestination, false);
 598
 599     announceInitiatorAbort(adaptedHeadSource);
 600
 601     if (adaptedBodySource != NULL)
 602         stopConsumingFrom(adaptedBodySource);
 603
 604     if (!adaptationAccessCheckPending) // we cannot cancel a pending callback
 605         assert(doneWithAdaptation()); // make sure the two methods are in sync
 606 }
 607
 608 bool
 609 ServerStateData::doneWithAdaptation() const
 610 {
 611     return !adaptationAccessCheckPending &&
 612            !virginBodyDestination && !adaptedHeadSource && !adaptedBodySource;
 613 }
 614
 615 // sends virgin reply body to ICAP, buffering excesses if needed
 616 void
 617 ServerStateData::adaptVirginReplyBody(const char *data, ssize_t len)
 618 {
 619     assert(startedAdaptation);
 620
 621     if (!virginBodyDestination) {
 622         debugs(11,3, HERE << "ICAP does not want more virgin body");
 623         return;
 624     }
 625
 626     // grow overflow area if already overflowed
 627     if (responseBodyBuffer) {
 628         responseBodyBuffer->append(data, len);
 629         data = responseBodyBuffer->content();
 630         len = responseBodyBuffer->contentSize();
 631     }
 632
 633     const ssize_t putSize = virginBodyDestination->putMoreData(data, len);
 634     data += putSize;
 635     len -= putSize;
 636
 637     // if we had overflow area, shrink it as necessary
 638     if (responseBodyBuffer) {
 639         if (putSize == responseBodyBuffer->contentSize()) {
 640             delete responseBodyBuffer;
 641             responseBodyBuffer = NULL;
 642         } else {
 643             responseBodyBuffer->consume(putSize);
 644         }
 645         return;
 646     }
 647
 648     // if we did not have an overflow area, create it as needed
 649     if (len > 0) {
 650         assert(!responseBodyBuffer);
 651         responseBodyBuffer = new MemBuf;
 652         responseBodyBuffer->init(4096, SQUID_TCP_SO_RCVBUF * 10);
 653         responseBodyBuffer->append(data, len);
 654     }
 655 }
 656
 657 // can supply more virgin response body data
 658 void
 659 ServerStateData::noteMoreBodySpaceAvailable(BodyPipe::Pointer)
 660 {
 661     if (responseBodyBuffer) {
 662         addVirginReplyBody(NULL, 0); // kick the buffered fragment alive again
 663         if (completed && !responseBodyBuffer) {
 664             serverComplete2();
 665             return;
 666         }
 667     }
 668     maybeReadVirginBody();
 669 }
 670
 671 // the consumer of our virgin response body aborted
 672 void
 673 ServerStateData::noteBodyConsumerAborted(BodyPipe::Pointer)
 674 {
 675     stopProducingFor(virginBodyDestination, false);
 676
 677     // do not force closeServer here in case we need to bypass AdaptationQueryAbort
 678
 679     if (doneWithAdaptation()) // we may still be receiving adapted response
 680         handleAdaptationCompleted();
 681 }
 682
 683 // received adapted response headers (body may follow)
 684 void
 685 ServerStateData::noteAdaptationAnswer(const Adaptation::Answer &answer)
 686 {
 687     clearAdaptation(adaptedHeadSource); // we do not expect more messages
 688
 689     switch (answer.kind) {
 690     case Adaptation::Answer::akForward:
 691         handleAdaptedHeader(const_cast<HttpMsg*>(answer.message.getRaw()));
 692         break;
 693
 694     case Adaptation::Answer::akBlock:
 695         handleAdaptationBlocked(answer);
 696         break;
 697
 698     case Adaptation::Answer::akError:
 699         handleAdaptationAborted(!answer.final);
 700         break;
 701     }
 702 }
 703
 704 void
 705 ServerStateData::handleAdaptedHeader(HttpMsg *msg)
 706 {
 707     if (abortOnBadEntry("entry went bad while waiting for adapted headers")) {
 708         // If the adapted response has a body, the ICAP side needs to know
 709         // that nobody will consume that body. We will be destroyed upon
 710         // return. Tell the ICAP side that it is on its own.
 711         HttpReply *rep = dynamic_cast<HttpReply*>(msg);
 712         assert(rep);
 713         if (rep->body_pipe != NULL)
 714             rep->body_pipe->expectNoConsumption();
 715
 716         return;
 717     }
 718
 719     HttpReply *rep = dynamic_cast<HttpReply*>(msg);
 720     assert(rep);
 721     debugs(11,5, HERE << this << " setting adapted reply to " << rep);
 722     setFinalReply(rep);
 723
 724     assert(!adaptedBodySource);
 725     if (rep->body_pipe != NULL) {
 726         // subscribe to receive adapted body
 727         adaptedBodySource = rep->body_pipe;
 728         // assume that ICAP does not auto-consume on failures
 729         const bool result = adaptedBodySource->setConsumerIfNotLate(this);
 730         assert(result);
 731     } else {
 732         // no body
 733         if (doneWithAdaptation()) // we may still be sending virgin response
 734             handleAdaptationCompleted();
 735     }
 736 }
 737
 738 void
 739 ServerStateData::resumeBodyStorage()
 740 {
 741     if (abortOnBadEntry("store entry aborted while kick producer callback"))
 742         return;
 743
 744     if (!adaptedBodySource)
 745         return;
 746
 747     handleMoreAdaptedBodyAvailable();
 748
 749     if (adaptedBodySource != NULL && adaptedBodySource->exhausted())
 750         endAdaptedBodyConsumption();
 751 }
 752
 753 // more adapted response body is available
 754 void
 755 ServerStateData::handleMoreAdaptedBodyAvailable()
 756 {
 757     if (abortOnBadEntry("entry refuses adapted body"))
 758         return;
 759
 760     assert(entry);
 761
 762     size_t contentSize = adaptedBodySource->buf().contentSize();
 763
 764     if (!contentSize)
 765         return; // XXX: bytesWanted asserts on zero-size ranges
 766
 767     const size_t spaceAvailable = entry->bytesWanted(Range<size_t>(0, contentSize), true);
 768
 769     if (spaceAvailable < contentSize ) {
 770         // No or partial body data consuming
 771         typedef NullaryMemFunT<ServerStateData> Dialer;
 772         AsyncCall::Pointer call = asyncCall(93, 5, "ServerStateData::resumeBodyStorage",
 773                                             Dialer(this, &ServerStateData::resumeBodyStorage));
 774         entry->deferProducer(call);
 775     }
 776
 777     if (!spaceAvailable)  {
 778         debugs(11, 5, HERE << "NOT storing " << contentSize << " bytes of adapted " <<
 779                "response body at offset " << adaptedBodySource->consumedSize());
 780         return;
 781     }
 782
 783     if (spaceAvailable < contentSize ) {
 784         debugs(11, 5, HERE << "postponing storage of " <<
 785                (contentSize - spaceAvailable) << " body bytes");
 786         contentSize = spaceAvailable;
 787     }
 788
 789     debugs(11,5, HERE << "storing " << contentSize << " bytes of adapted " <<
 790            "response body at offset " << adaptedBodySource->consumedSize());
 791
 792     BodyPipeCheckout bpc(*adaptedBodySource);
 793     const StoreIOBuffer ioBuf(&bpc.buf, currentOffset, contentSize);
 794     currentOffset += ioBuf.length;
 795     entry->write(ioBuf);
 796     bpc.buf.consume(contentSize);
 797     bpc.checkIn();
 798 }
 799
 800 // the entire adapted response body was produced, successfully
 801 void
 802 ServerStateData::handleAdaptedBodyProductionEnded()
 803 {
 804     if (abortOnBadEntry("entry went bad while waiting for adapted body eof"))
 805         return;
 806
 807     // end consumption if we consumed everything
 808     if (adaptedBodySource != NULL && adaptedBodySource->exhausted())
 809         endAdaptedBodyConsumption();
 810     // else resumeBodyStorage() will eventually consume the rest
 811 }
 812
 813 void
 814 ServerStateData::endAdaptedBodyConsumption()
 815 {
 816     stopConsumingFrom(adaptedBodySource);
 817     handleAdaptationCompleted();
 818 }
 819
 820 // premature end of the adapted response body
 821 void ServerStateData::handleAdaptedBodyProducerAborted()
 822 {
 823     stopConsumingFrom(adaptedBodySource);
 824     handleAdaptationAborted();
 825 }
 826
 827 // common part of noteAdaptationAnswer and handleAdaptedBodyProductionEnded
 828 void
 829 ServerStateData::handleAdaptationCompleted()
 830 {
 831     debugs(11,5, HERE << "handleAdaptationCompleted");
 832     cleanAdaptation();
 833
 834     // We stop reading origin response because we have no place to put it and
 835     // cannot use it. If some origin servers do not like that or if we want to
 836     // reuse more pconns, we can add code to discard unneeded origin responses.
 837     if (!doneWithServer()) {
 838         debugs(11,3, HERE << "closing origin conn due to ICAP completion");
 839         closeServer();
 840     }
 841
 842     completeForwarding();
 843 }
 844
 845 // common part of noteAdaptation*Aborted and noteBodyConsumerAborted methods
 846 void
 847 ServerStateData::handleAdaptationAborted(bool bypassable)
 848 {
 849     debugs(11,5, HERE << "handleAdaptationAborted; bypassable: " << bypassable <<
 850            ", entry empty: " << entry->isEmpty());
 851
 852     if (abortOnBadEntry("entry went bad while ICAP aborted"))
 853         return;
 854
 855     // TODO: bypass if possible
 856
 857     if (entry->isEmpty()) {
 858         debugs(11,9, HERE << "creating ICAP error entry after ICAP failure");
 859         ErrorState *err = new ErrorState(ERR_ICAP_FAILURE, Http::scInternalServerError, request);
 860         err->detailError(ERR_DETAIL_ICAP_RESPMOD_EARLY);
 861         fwd->fail(err);
 862         fwd->dontRetry(true);
 863     } else if (request) { // update logged info directly
 864         request->detailError(ERR_ICAP_FAILURE, ERR_DETAIL_ICAP_RESPMOD_LATE);
 865     }
 866
 867     abortTransaction("ICAP failure");
 868 }
 869
 870 // adaptation service wants us to deny HTTP client access to this response
 871 void
 872 ServerStateData::handleAdaptationBlocked(const Adaptation::Answer &answer)
 873 {
 874     debugs(11,5, HERE << answer.ruleId);
 875
 876     if (abortOnBadEntry("entry went bad while ICAP aborted"))
 877         return;
 878
 879     if (!entry->isEmpty()) { // too late to block (should not really happen)
 880         if (request)
 881             request->detailError(ERR_ICAP_FAILURE, ERR_DETAIL_RESPMOD_BLOCK_LATE);
 882         abortTransaction("late adaptation block");
 883         return;
 884     }
 885
 886     debugs(11,7, HERE << "creating adaptation block response");
 887
 888     err_type page_id =
 889         aclGetDenyInfoPage(&Config.denyInfoList, answer.ruleId.termedBuf(), 1);
 890     if (page_id == ERR_NONE)
 891         page_id = ERR_ACCESS_DENIED;
 892
 893     ErrorState *err = new ErrorState(page_id, Http::scForbidden, request);
 894     err->detailError(ERR_DETAIL_RESPMOD_BLOCK_EARLY);
 895     fwd->fail(err);
 896     fwd->dontRetry(true);
 897
 898     abortTransaction("timely adaptation block");
 899 }
 900
 901 void
 902 ServerStateData::noteAdaptationAclCheckDone(Adaptation::ServiceGroupPointer group)
 903 {
 904     adaptationAccessCheckPending = false;
 905
 906     if (abortOnBadEntry("entry went bad while waiting for ICAP ACL check"))
 907         return;
 908
 909     // TODO: Should nonICAP and postICAP path check this on the server-side?
 910     // That check now only happens on client-side, in processReplyAccess().
 911     if (virginReply()->expectedBodyTooLarge(*request)) {
 912         sendBodyIsTooLargeError();
 913         return;
 914     }
 915     // TODO: Should we check receivedBodyTooLarge on the server-side as well?
 916
 917     if (!group) {
 918         debugs(11,3, HERE << "no adapation needed");
 919         setFinalReply(virginReply());
 920         processReplyBody();
 921         return;
 922     }
 923
 924     startAdaptation(group, originalRequest());
 925     processReplyBody();
 926 }
 927 #endif
 928
 929 void
 930 ServerStateData::sendBodyIsTooLargeError()
 931 {
 932     ErrorState *err = new ErrorState(ERR_TOO_BIG, Http::scForbidden, request);
 933     fwd->fail(err);
 934     fwd->dontRetry(true);
 935     abortTransaction("Virgin body too large.");
 936 }
 937
 938 // TODO: when HttpStateData sends all errors to ICAP,
 939 // we should be able to move this at the end of setVirginReply().
 940 void
 941 ServerStateData::adaptOrFinalizeReply()
 942 {
 943 #if USE_ADAPTATION
 944     // TODO: merge with client side and return void to hide the on/off logic?
 945     // The callback can be called with a NULL service if adaptation is off.
 946     adaptationAccessCheckPending = Adaptation::AccessCheck::Start(
 947                                        Adaptation::methodRespmod, Adaptation::pointPreCache,
 948                                        originalRequest(), virginReply(), fwd->al, this);
 949     debugs(11,5, HERE << "adaptationAccessCheckPending=" << adaptationAccessCheckPending);
 950     if (adaptationAccessCheckPending)
 951         return;
 952 #endif
 953
 954     setFinalReply(virginReply());
 955 }
 956
 957 /// initializes bodyBytesRead stats if needed and applies delta
 958 void
 959 ServerStateData::adjustBodyBytesRead(const int64_t delta)
 960 {
 961     int64_t &bodyBytesRead = originalRequest()->hier.bodyBytesRead;
 962
 963     // if we got here, do not log a dash even if we got nothing from the server
 964     if (bodyBytesRead < 0)
 965         bodyBytesRead = 0;
 966
 967     bodyBytesRead += delta; // supports negative and zero deltas
 968
 969     // check for overflows ("infinite" response?) and undeflows (a bug)
 970     Must(bodyBytesRead >= 0);
 971 }
 972
 973 void
 974 ServerStateData::addVirginReplyBody(const char *data, ssize_t len)
 975 {
 976     adjustBodyBytesRead(len);
 977
 978 #if USE_ADAPTATION
 979     assert(!adaptationAccessCheckPending); // or would need to buffer while waiting
 980     if (startedAdaptation) {
 981         adaptVirginReplyBody(data, len);
 982         return;
 983     }
 984 #endif
 985     storeReplyBody(data, len);
 986 }
 987
 988 // writes virgin or adapted reply body to store
 989 void
 990 ServerStateData::storeReplyBody(const char *data, ssize_t len)
 991 {
 992     // write even if len is zero to push headers towards the client side
 993     entry->write (StoreIOBuffer(len, currentOffset, (char*)data));
 994
 995     currentOffset += len;
 996 }
 997
 998 size_t ServerStateData::replyBodySpace(const MemBuf &readBuf,
 999                                        const size_t minSpace) const
1000 {
1001     size_t space = readBuf.spaceSize(); // available space w/o heroic measures
1002     if (space < minSpace) {
1003         const size_t maxSpace = readBuf.potentialSpaceSize(); // absolute best
1004         space = min(minSpace, maxSpace); // do not promise more than asked
1005     }
1006
1007 #if USE_ADAPTATION
1008     if (responseBodyBuffer) {
1009         return 0;       // Stop reading if already overflowed waiting for ICAP to catch up
1010     }
1011
1012     if (virginBodyDestination != NULL) {
1013         /*
1014          * BodyPipe buffer has a finite size limit.  We
1015          * should not read more data from the network than will fit
1016          * into the pipe buffer or we _lose_ what did not fit if
1017          * the response ends sooner that BodyPipe frees up space:
1018          * There is no code to keep pumping data into the pipe once
1019          * response ends and serverComplete() is called.
1020          *
1021          * If the pipe is totally full, don't register the read handler.
1022          * The BodyPipe will call our noteMoreBodySpaceAvailable() method
1023          * when it has free space again.
1024          */
1025         size_t adaptation_space =
1026             virginBodyDestination->buf().potentialSpaceSize();
1027
1028         debugs(11,9, "ServerStateData may read up to min(" <<
1029                adaptation_space << ", " << space << ") bytes");
1030
1031         if (adaptation_space < space)
1032             space = adaptation_space;
1033     }
1034 #endif
1035
1036     return space;
1037 }