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