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