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     entry->replaceHttpReply(theFinalReply);
 167     haveParsedReplyHeaders();
 168
 169     return theFinalReply;
 170 }
 171
 172 // called when no more server communication is expected; may quit
 173 void
 174 ServerStateData::serverComplete()
 175 {
 176     debugs(11,5,HERE << "serverComplete " << this);
 177
 178     if (!doneWithServer()) {
 179         closeServer();
 180         assert(doneWithServer());
 181     }
 182
 183     completed = true;
 184
 185     HttpRequest *r = originalRequest();
 186     r->hier.total_response_time = r->hier.first_conn_start.tv_sec ?
 187                                   tvSubMsec(r->hier.first_conn_start, current_time) : -1;
 188
 189     if (requestBodySource != NULL)
 190         stopConsumingFrom(requestBodySource);
 191
 192     if (responseBodyBuffer != NULL)
 193         return;
 194
 195     serverComplete2();
 196 }
 197
 198 void
 199 ServerStateData::serverComplete2()
 200 {
 201     debugs(11,5,HERE << "serverComplete2 " << this);
 202
 203 #if USE_ADAPTATION
 204     if (virginBodyDestination != NULL)
 205         stopProducingFor(virginBodyDestination, true);
 206
 207     if (!doneWithAdaptation())
 208         return;
 209 #endif
 210
 211     completeForwarding();
 212     quitIfAllDone();
 213 }
 214
 215 // When we are done talking to the primary server, we may be still talking
 216 // to the ICAP service. And vice versa. Here, we quit only if we are done
 217 // talking to both.
 218 void ServerStateData::quitIfAllDone()
 219 {
 220 #if USE_ADAPTATION
 221     if (!doneWithAdaptation()) {
 222         debugs(11,5, HERE << "transaction not done: still talking to ICAP");
 223         return;
 224     }
 225 #endif
 226
 227     if (!doneWithServer()) {
 228         debugs(11,5, HERE << "transaction not done: still talking to server");
 229         return;
 230     }
 231
 232     debugs(11,3, HERE << "transaction done");
 233
 234     deleteThis("ServerStateData::quitIfAllDone");
 235 }
 236
 237 // FTP side overloads this to work around multiple calls to fwd->complete
 238 void
 239 ServerStateData::completeForwarding()
 240 {
 241     debugs(11,5, HERE << "completing forwarding for "  << fwd);
 242     assert(fwd != NULL);
 243     fwd->complete();
 244 }
 245
 246 // Register to receive request body
 247 bool ServerStateData::startRequestBodyFlow()
 248 {
 249     HttpRequest *r = originalRequest();
 250     assert(r->body_pipe != NULL);
 251     requestBodySource = r->body_pipe;
 252     if (requestBodySource->setConsumerIfNotLate(this)) {
 253         debugs(11,3, HERE << "expecting request body from " <<
 254                requestBodySource->status());
 255         return true;
 256     }
 257
 258     debugs(11,3, HERE << "aborting on partially consumed request body: " <<
 259            requestBodySource->status());
 260     requestBodySource = NULL;
 261     return false;
 262 }
 263
 264 // Entry-dependent callbacks use this check to quit if the entry went bad
 265 bool
 266 ServerStateData::abortOnBadEntry(const char *abortReason)
 267 {
 268     if (entry->isAccepting())
 269         return false;
 270
 271     debugs(11,5, HERE << "entry is not Accepting!");
 272     abortTransaction(abortReason);
 273     return true;
 274 }
 275
 276 // more request or adapted response body is available
 277 void
 278 ServerStateData::noteMoreBodyDataAvailable(BodyPipe::Pointer bp)
 279 {
 280 #if USE_ADAPTATION
 281     if (adaptedBodySource == bp) {
 282         handleMoreAdaptedBodyAvailable();
 283         return;
 284     }
 285 #endif
 286     handleMoreRequestBodyAvailable();
 287 }
 288
 289 // the entire request or adapted response body was provided, successfully
 290 void
 291 ServerStateData::noteBodyProductionEnded(BodyPipe::Pointer bp)
 292 {
 293 #if USE_ADAPTATION
 294     if (adaptedBodySource == bp) {
 295         handleAdaptedBodyProductionEnded();
 296         return;
 297     }
 298 #endif
 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     handleRequestBodyProducerAborted();
 313 }
 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, 1, "sentRequestBody error: FD " << io.fd << ": " << xstrerr(errno));
 386         ErrorState *err;
 387         err = errorCon(ERR_WRITE_ERROR, HTTP_BAD_GATEWAY, fwd->request);
 388         err->xerrno = errno;
 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 (usually by kids) when we have final (possibly adapted) reply headers
 524 void
 525 ServerStateData::haveParsedReplyHeaders()
 526 {
 527     Must(theFinalReply);
 528     maybePurgeOthers();
 529 }
 530
 531 HttpRequest *
 532 ServerStateData::originalRequest()
 533 {
 534     return request;
 535 }
 536
 537 #if USE_ADAPTATION
 538 /// Initiate an asynchronous adaptation transaction which will call us back.
 539 void
 540 ServerStateData::startAdaptation(const Adaptation::ServiceGroupPointer &group, HttpRequest *cause)
 541 {
 542     debugs(11, 5, "ServerStateData::startAdaptation() called");
 543     // check whether we should be sending a body as well
 544     // start body pipe to feed ICAP transaction if needed
 545     assert(!virginBodyDestination);
 546     HttpReply *vrep = virginReply();
 547     assert(!vrep->body_pipe);
 548     int64_t size = 0;
 549     if (vrep->expectingBody(cause->method, size) && size) {
 550         virginBodyDestination = new BodyPipe(this);
 551         vrep->body_pipe = virginBodyDestination;
 552         debugs(93, 6, HERE << "will send virgin reply body to " <<
 553                virginBodyDestination << "; size: " << size);
 554         if (size > 0)
 555             virginBodyDestination->setBodySize(size);
 556     }
 557
 558     adaptedHeadSource = initiateAdaptation(
 559                             new Adaptation::Iterator(vrep, cause, group));
 560     startedAdaptation = initiated(adaptedHeadSource);
 561     Must(startedAdaptation);
 562 }
 563
 564 // properly cleans up ICAP-related state
 565 // may be called multiple times
 566 void ServerStateData::cleanAdaptation()
 567 {
 568     debugs(11,5, HERE << "cleaning ICAP; ACL: " << adaptationAccessCheckPending);
 569
 570     if (virginBodyDestination != NULL)
 571         stopProducingFor(virginBodyDestination, false);
 572
 573     announceInitiatorAbort(adaptedHeadSource);
 574
 575     if (adaptedBodySource != NULL)
 576         stopConsumingFrom(adaptedBodySource);
 577
 578     if (!adaptationAccessCheckPending) // we cannot cancel a pending callback
 579         assert(doneWithAdaptation()); // make sure the two methods are in sync
 580 }
 581
 582 bool
 583 ServerStateData::doneWithAdaptation() const
 584 {
 585     return !adaptationAccessCheckPending &&
 586            !virginBodyDestination && !adaptedHeadSource && !adaptedBodySource;
 587 }
 588
 589 // sends virgin reply body to ICAP, buffering excesses if needed
 590 void
 591 ServerStateData::adaptVirginReplyBody(const char *data, ssize_t len)
 592 {
 593     assert(startedAdaptation);
 594
 595     if (!virginBodyDestination) {
 596         debugs(11,3, HERE << "ICAP does not want more virgin body");
 597         return;
 598     }
 599
 600     // grow overflow area if already overflowed
 601     if (responseBodyBuffer) {
 602         responseBodyBuffer->append(data, len);
 603         data = responseBodyBuffer->content();
 604         len = responseBodyBuffer->contentSize();
 605     }
 606
 607     const ssize_t putSize = virginBodyDestination->putMoreData(data, len);
 608     data += putSize;
 609     len -= putSize;
 610
 611     // if we had overflow area, shrink it as necessary
 612     if (responseBodyBuffer) {
 613         if (putSize == responseBodyBuffer->contentSize()) {
 614             delete responseBodyBuffer;
 615             responseBodyBuffer = NULL;
 616         } else {
 617             responseBodyBuffer->consume(putSize);
 618         }
 619         return;
 620     }
 621
 622     // if we did not have an overflow area, create it as needed
 623     if (len > 0) {
 624         assert(!responseBodyBuffer);
 625         responseBodyBuffer = new MemBuf;
 626         responseBodyBuffer->init(4096, SQUID_TCP_SO_RCVBUF * 10);
 627         responseBodyBuffer->append(data, len);
 628     }
 629 }
 630
 631 // can supply more virgin response body data
 632 void
 633 ServerStateData::noteMoreBodySpaceAvailable(BodyPipe::Pointer)
 634 {
 635     if (responseBodyBuffer) {
 636         addVirginReplyBody(NULL, 0); // kick the buffered fragment alive again
 637         if (completed && !responseBodyBuffer) {
 638             serverComplete2();
 639             return;
 640         }
 641     }
 642     maybeReadVirginBody();
 643 }
 644
 645 // the consumer of our virgin response body aborted
 646 void
 647 ServerStateData::noteBodyConsumerAborted(BodyPipe::Pointer)
 648 {
 649     stopProducingFor(virginBodyDestination, false);
 650
 651     // do not force closeServer here in case we need to bypass AdaptationQueryAbort
 652
 653     if (doneWithAdaptation()) // we may still be receiving adapted response
 654         handleAdaptationCompleted();
 655 }
 656
 657 // received adapted response headers (body may follow)
 658 void
 659 ServerStateData::noteAdaptationAnswer(const Adaptation::Answer &answer)
 660 {
 661     clearAdaptation(adaptedHeadSource); // we do not expect more messages
 662
 663     switch (answer.kind) {
 664     case Adaptation::Answer::akForward:
 665         handleAdaptedHeader(answer.message);
 666         break;
 667
 668     case Adaptation::Answer::akBlock:
 669         handleAdaptationBlocked(answer);
 670         break;
 671
 672     case Adaptation::Answer::akError:
 673         handleAdaptationAborted(!answer.final);
 674         break;
 675     }
 676 }
 677
 678 void
 679 ServerStateData::handleAdaptedHeader(HttpMsg *msg)
 680 {
 681     if (abortOnBadEntry("entry went bad while waiting for adapted headers"))
 682         return;
 683
 684     HttpReply *rep = dynamic_cast<HttpReply*>(msg);
 685     assert(rep);
 686     debugs(11,5, HERE << this << " setting adapted reply to " << rep);
 687     setFinalReply(rep);
 688
 689     assert(!adaptedBodySource);
 690     if (rep->body_pipe != NULL) {
 691         // subscribe to receive adapted body
 692         adaptedBodySource = rep->body_pipe;
 693         // assume that ICAP does not auto-consume on failures
 694         assert(adaptedBodySource->setConsumerIfNotLate(this));
 695     } else {
 696         // no body
 697         if (doneWithAdaptation()) // we may still be sending virgin response
 698             handleAdaptationCompleted();
 699     }
 700 }
 701
 702 // more adapted response body is available
 703 void
 704 ServerStateData::handleMoreAdaptedBodyAvailable()
 705 {
 706     const size_t contentSize = adaptedBodySource->buf().contentSize();
 707
 708     debugs(11,5, HERE << "consuming " << contentSize << " bytes of adapted " <<
 709            "response body at offset " << adaptedBodySource->consumedSize());
 710
 711     if (abortOnBadEntry("entry refuses adapted body"))
 712         return;
 713
 714     assert(entry);
 715     BodyPipeCheckout bpc(*adaptedBodySource);
 716     const StoreIOBuffer ioBuf(&bpc.buf, currentOffset);
 717     currentOffset += bpc.buf.size;
 718     entry->write(ioBuf);
 719     bpc.buf.consume(contentSize);
 720     bpc.checkIn();
 721 }
 722
 723 // the entire adapted response body was produced, successfully
 724 void
 725 ServerStateData::handleAdaptedBodyProductionEnded()
 726 {
 727     stopConsumingFrom(adaptedBodySource);
 728
 729     if (abortOnBadEntry("entry went bad while waiting for adapted body eof"))
 730         return;
 731
 732     handleAdaptationCompleted();
 733 }
 734
 735 // premature end of the adapted response body
 736 void ServerStateData::handleAdaptedBodyProducerAborted()
 737 {
 738     stopConsumingFrom(adaptedBodySource);
 739     handleAdaptationAborted();
 740 }
 741
 742 // common part of noteAdaptationAnswer and handleAdaptedBodyProductionEnded
 743 void
 744 ServerStateData::handleAdaptationCompleted()
 745 {
 746     debugs(11,5, HERE << "handleAdaptationCompleted");
 747     cleanAdaptation();
 748
 749     // We stop reading origin response because we have no place to put it and
 750     // cannot use it. If some origin servers do not like that or if we want to
 751     // reuse more pconns, we can add code to discard unneeded origin responses.
 752     if (!doneWithServer()) {
 753         debugs(11,3, HERE << "closing origin conn due to ICAP completion");
 754         closeServer();
 755     }
 756
 757     completeForwarding();
 758     quitIfAllDone();
 759 }
 760
 761
 762 // common part of noteAdaptation*Aborted and noteBodyConsumerAborted methods
 763 void
 764 ServerStateData::handleAdaptationAborted(bool bypassable)
 765 {
 766     debugs(11,5, HERE << "handleAdaptationAborted; bypassable: " << bypassable <<
 767            ", entry empty: " << entry->isEmpty());
 768
 769     if (abortOnBadEntry("entry went bad while ICAP aborted"))
 770         return;
 771
 772     // TODO: bypass if possible
 773
 774     if (entry->isEmpty()) {
 775         debugs(11,9, HERE << "creating ICAP error entry after ICAP failure");
 776         ErrorState *err = errorCon(ERR_ICAP_FAILURE, HTTP_INTERNAL_SERVER_ERROR, request);
 777         err->xerrno = ERR_DETAIL_ICAP_RESPMOD_EARLY;
 778         fwd->fail(err);
 779         fwd->dontRetry(true);
 780     } else if (request) { // update logged info directly
 781         request->detailError(ERR_ICAP_FAILURE, ERR_DETAIL_ICAP_RESPMOD_LATE);
 782     }
 783
 784     abortTransaction("ICAP failure");
 785 }
 786
 787 // adaptation service wants us to deny HTTP client access to this response
 788 void
 789 ServerStateData::handleAdaptationBlocked(const Adaptation::Answer &answer)
 790 {
 791     debugs(11,5, HERE << answer.ruleId);
 792
 793     if (abortOnBadEntry("entry went bad while ICAP aborted"))
 794         return;
 795
 796     if (!entry->isEmpty()) { // too late to block (should not really happen)
 797         if (request)
 798             request->detailError(ERR_ICAP_FAILURE, ERR_DETAIL_RESPMOD_BLOCK_LATE);
 799         abortTransaction("late adaptation block");
 800         return;
 801     }
 802
 803     debugs(11,7, HERE << "creating adaptation block response");
 804
 805     err_type page_id =
 806         aclGetDenyInfoPage(&Config.denyInfoList, answer.ruleId.termedBuf(), 1);
 807     if (page_id == ERR_NONE)
 808         page_id = ERR_ACCESS_DENIED;
 809
 810     ErrorState *err = errorCon(page_id, HTTP_FORBIDDEN, request);
 811     err->xerrno = ERR_DETAIL_RESPMOD_BLOCK_EARLY;
 812     fwd->fail(err);
 813     fwd->dontRetry(true);
 814
 815     abortTransaction("timely adaptation block");
 816 }
 817
 818 void
 819 ServerStateData::adaptationAclCheckDone(Adaptation::ServiceGroupPointer group)
 820 {
 821     adaptationAccessCheckPending = false;
 822
 823     if (abortOnBadEntry("entry went bad while waiting for ICAP ACL check"))
 824         return;
 825
 826     // TODO: Should nonICAP and postICAP path check this on the server-side?
 827     // That check now only happens on client-side, in processReplyAccess().
 828     if (virginReply()->expectedBodyTooLarge(*request)) {
 829         sendBodyIsTooLargeError();
 830         return;
 831     }
 832     // TODO: Should we check receivedBodyTooLarge on the server-side as well?
 833
 834     if (!group) {
 835         debugs(11,3, HERE << "no adapation needed");
 836         setFinalReply(virginReply());
 837         processReplyBody();
 838         return;
 839     }
 840
 841     startAdaptation(group, originalRequest());
 842     processReplyBody();
 843 }
 844
 845 void
 846 ServerStateData::adaptationAclCheckDoneWrapper(Adaptation::ServiceGroupPointer group, void *data)
 847 {
 848     ServerStateData *state = (ServerStateData *)data;
 849     state->adaptationAclCheckDone(group);
 850 }
 851 #endif
 852
 853 void
 854 ServerStateData::sendBodyIsTooLargeError()
 855 {
 856     ErrorState *err = errorCon(ERR_TOO_BIG, HTTP_FORBIDDEN, request);
 857     err->xerrno = errno;
 858     fwd->fail(err);
 859     fwd->dontRetry(true);
 860     abortTransaction("Virgin body too large.");
 861 }
 862
 863 // TODO: when HttpStateData sends all errors to ICAP,
 864 // we should be able to move this at the end of setVirginReply().
 865 void
 866 ServerStateData::adaptOrFinalizeReply()
 867 {
 868 #if USE_ADAPTATION
 869     // TODO: merge with client side and return void to hide the on/off logic?
 870     // The callback can be called with a NULL service if adaptation is off.
 871     adaptationAccessCheckPending = Adaptation::AccessCheck::Start(
 872                                        Adaptation::methodRespmod, Adaptation::pointPreCache,
 873                                        originalRequest(), virginReply(), adaptationAclCheckDoneWrapper, this);
 874     debugs(11,5, HERE << "adaptationAccessCheckPending=" << adaptationAccessCheckPending);
 875     if (adaptationAccessCheckPending)
 876         return;
 877 #endif
 878
 879     setFinalReply(virginReply());
 880 }
 881
 882 /// initializes bodyBytesRead stats if needed and applies delta
 883 void
 884 ServerStateData::adjustBodyBytesRead(const int64_t delta)
 885 {
 886     int64_t &bodyBytesRead = originalRequest()->hier.bodyBytesRead;
 887
 888     // if we got here, do not log a dash even if we got nothing from the server
 889     if (bodyBytesRead < 0)
 890         bodyBytesRead = 0;
 891
 892     bodyBytesRead += delta; // supports negative and zero deltas
 893
 894     // check for overflows ("infinite" response?) and undeflows (a bug)
 895     Must(bodyBytesRead >= 0);
 896 }
 897
 898 void
 899 ServerStateData::addVirginReplyBody(const char *data, ssize_t len)
 900 {
 901     adjustBodyBytesRead(len);
 902
 903 #if USE_ADAPTATION
 904     assert(!adaptationAccessCheckPending); // or would need to buffer while waiting
 905     if (startedAdaptation) {
 906         adaptVirginReplyBody(data, len);
 907         return;
 908     }
 909 #endif
 910     storeReplyBody(data, len);
 911 }
 912
 913 // writes virgin or adapted reply body to store
 914 void
 915 ServerStateData::storeReplyBody(const char *data, ssize_t len)
 916 {
 917     // write even if len is zero to push headers towards the client side
 918     entry->write (StoreIOBuffer(len, currentOffset, (char*)data));
 919
 920     currentOffset += len;
 921 }
 922
 923 size_t ServerStateData::replyBodySpace(const MemBuf &readBuf,
 924                                        const size_t minSpace) const
 925 {
 926     size_t space = readBuf.spaceSize(); // available space w/o heroic measures
 927     if (space < minSpace) {
 928         const size_t maxSpace = readBuf.potentialSpaceSize(); // absolute best
 929         space = min(minSpace, maxSpace); // do not promise more than asked
 930     }
 931
 932 #if USE_ADAPTATION
 933     if (responseBodyBuffer) {
 934         return 0;       // Stop reading if already overflowed waiting for ICAP to catch up
 935     }
 936
 937     if (virginBodyDestination != NULL) {
 938         /*
 939          * BodyPipe buffer has a finite size limit.  We
 940          * should not read more data from the network than will fit
 941          * into the pipe buffer or we _lose_ what did not fit if
 942          * the response ends sooner that BodyPipe frees up space:
 943          * There is no code to keep pumping data into the pipe once
 944          * response ends and serverComplete() is called.
 945          *
 946          * If the pipe is totally full, don't register the read handler.
 947          * The BodyPipe will call our noteMoreBodySpaceAvailable() method
 948          * when it has free space again.
 949          */
 950         size_t adaptation_space =
 951             virginBodyDestination->buf().potentialSpaceSize();
 952
 953         debugs(11,9, "ServerStateData may read up to min(" <<
 954                adaptation_space << ", " << space << ") bytes");
 955
 956         if (adaptation_space < space)
 957             space = adaptation_space;
 958     }
 959 #endif
 960
 961     return space;
 962 }