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