src/Server.cc

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