src/store_client.cc

   1
   2 /*
   3  * DEBUG: section 90    Storage Manager Client-Side Interface
   4  * AUTHOR: Duane Wessels
   5  *
   6  * SQUID Web Proxy Cache          http://www.squid-cache.org/
   7  * ----------------------------------------------------------
   8  *
   9  *  Squid is the result of efforts by numerous individuals from
  10  *  the Internet community; see the CONTRIBUTORS file for full
  11  *  details.   Many organizations have provided support for Squid's
  12  *  development; see the SPONSORS file for full details.  Squid is
  13  *  Copyrighted (C) 2001 by the Regents of the University of
  14  *  California; see the COPYRIGHT file for full details.  Squid
  15  *  incorporates software developed and/or copyrighted by other
  16  *  sources; see the CREDITS file for full details.
  17  *
  18  *  This program is free software; you can redistribute it and/or modify
  19  *  it under the terms of the GNU General Public License as published by
  20  *  the Free Software Foundation; either version 2 of the License, or
  21  *  (at your option) any later version.
  22  *
  23  *  This program is distributed in the hope that it will be useful,
  24  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  25  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  26  *  GNU General Public License for more details.
  27  *
  28  *  You should have received a copy of the GNU General Public License
  29  *  along with this program; if not, write to the Free Software
  30  *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111, USA.
  31  *
  32  * Portions copyright (c) 2003 Robert Collins <robertc@squid-cache.org>
  33  */
  34
  35 #include "squid.h"
  36 #include "event.h"
  37 #include "HttpReply.h"
  38 #include "HttpRequest.h"
  39 #include "MemBuf.h"
  40 #include "MemObject.h"
  41 #include "mime_header.h"
  42 #include "profiler/Profiler.h"
  43 #include "SquidConfig.h"
  44 #include "StatCounters.h"
  45 #include "StoreClient.h"
  46 #include "Store.h"
  47 #include "store_swapin.h"
  48 #include "StoreMeta.h"
  49 #include "StoreMetaUnpacker.h"
  50 #if USE_DELAY_POOLS
  51 #include "DelayPools.h"
  52 #endif
  53
  54 /*
  55  * NOTE: 'Header' refers to the swapfile metadata header.
  56  *       'OBJHeader' refers to the object header, with cannonical
  57  *       processed object headers (which may derive from FTP/HTTP etc
  58  *       upstream protocols
  59  *       'Body' refers to the swapfile body, which is the full
  60  *        HTTP reply (including HTTP headers and body).
  61  */
  62 static StoreIOState::STRCB storeClientReadBody;
  63 static StoreIOState::STRCB storeClientReadHeader;
  64 static void storeClientCopy2(StoreEntry * e, store_client * sc);
  65 static EVH storeClientCopyEvent;
  66 static bool CheckQuickAbortIsReasonable(StoreEntry * entry);
  67 static void CheckQuickAbort(StoreEntry * entry);
  68
  69 CBDATA_CLASS_INIT(store_client);
  70
  71 void *
  72 store_client::operator new (size_t)
  73 {
  74     CBDATA_INIT_TYPE(store_client);
  75     store_client *result = cbdataAlloc(store_client);
  76     return result;
  77 }
  78
  79 void
  80 store_client::operator delete (void *address)
  81 {
  82     store_client *t = static_cast<store_client *>(address);
  83     cbdataFree(t);
  84 }
  85
  86 bool
  87 store_client::memReaderHasLowerOffset(int64_t anOffset) const
  88 {
  89     return getType() == STORE_MEM_CLIENT && copyInto.offset < anOffset;
  90 }
  91
  92 int
  93 store_client::getType() const
  94 {
  95     return type;
  96 }
  97
  98 #if STORE_CLIENT_LIST_DEBUG
  99 static store_client *
 100 storeClientListSearch(const MemObject * mem, void *data)
 101 {
 102     dlink_node *node;
 103     store_client *sc = NULL;
 104
 105     for (node = mem->clients.head; node; node = node->next) {
 106         sc = node->data;
 107
 108         if (sc->owner == data)
 109             return sc;
 110     }
 111
 112     return NULL;
 113 }
 114
 115 int
 116 storeClientIsThisAClient(store_client * sc, void *someClient)
 117 {
 118     return sc->owner == someClient;
 119 }
 120
 121 #endif
 122 #include "HttpRequest.h"
 123
 124 /* add client with fd to client list */
 125 store_client *
 126 storeClientListAdd(StoreEntry * e, void *data)
 127 {
 128     MemObject *mem = e->mem_obj;
 129     store_client *sc;
 130     assert(mem);
 131 #if STORE_CLIENT_LIST_DEBUG
 132
 133     if (storeClientListSearch(mem, data) != NULL)
 134         /* XXX die! */
 135         assert(1 == 0);
 136
 137 #endif
 138
 139     sc = new store_client (e);
 140
 141     mem->addClient(sc);
 142
 143     return sc;
 144 }
 145
 146 void
 147 store_client::callback(ssize_t sz, bool error)
 148 {
 149     StoreIOBuffer result(sz, 0 ,copyInto.data);
 150
 151     if (sz < 0) {
 152         result.flags.error = 1;
 153         result.length = 0;
 154     } else {
 155         result.flags.error = error ? 1 : 0;
 156     }
 157
 158     result.offset = cmp_offset;
 159     assert(_callback.pending());
 160     cmp_offset = copyInto.offset + sz;
 161     STCB *temphandler = _callback.callback_handler;
 162     void *cbdata = _callback.callback_data;
 163     _callback = Callback(NULL, NULL);
 164     copyInto.data = NULL;
 165
 166     if (cbdataReferenceValid(cbdata))
 167         temphandler(cbdata, result);
 168
 169     cbdataReferenceDone(cbdata);
 170 }
 171
 172 static void
 173 storeClientCopyEvent(void *data)
 174 {
 175     store_client *sc = (store_client *)data;
 176     debugs(90, 3, "storeClientCopyEvent: Running");
 177     assert (sc->flags.copy_event_pending);
 178     sc->flags.copy_event_pending = false;
 179
 180     if (!sc->_callback.pending())
 181         return;
 182
 183     storeClientCopy2(sc->entry, sc);
 184 }
 185
 186 store_client::store_client(StoreEntry *e) : entry (e)
 187 #if USE_DELAY_POOLS
 188         , delayId()
 189 #endif
 190         , type (e->storeClientType())
 191         ,  object_ok(true)
 192 {
 193     cmp_offset = 0;
 194     flags.disk_io_pending = false;
 195     ++ entry->refcount;
 196
 197     if (getType() == STORE_DISK_CLIENT)
 198         /* assert we'll be able to get the data we want */
 199         /* maybe we should open swapin_sio here */
 200         assert(entry->swap_filen > -1 || entry->swappingOut());
 201
 202 #if STORE_CLIENT_LIST_DEBUG
 203
 204     owner = cbdataReference(data);
 205
 206 #endif
 207 }
 208
 209 store_client::~store_client()
 210 {}
 211
 212 /* copy bytes requested by the client */
 213 void
 214 storeClientCopy(store_client * sc,
 215                 StoreEntry * e,
 216                 StoreIOBuffer copyInto,
 217                 STCB * callback,
 218                 void *data)
 219 {
 220     assert (sc != NULL);
 221     sc->copy(e, copyInto,callback,data);
 222 }
 223
 224 void
 225 store_client::copy(StoreEntry * anEntry,
 226                    StoreIOBuffer copyRequest,
 227                    STCB * callback_fn,
 228                    void *data)
 229 {
 230     assert (anEntry == entry);
 231     assert (callback_fn);
 232     assert (data);
 233     assert(!EBIT_TEST(entry->flags, ENTRY_ABORTED));
 234     debugs(90, 3, "store_client::copy: " << entry->getMD5Text() << ", from " <<
 235            copyRequest.offset << ", for length " <<
 236            (int) copyRequest.length << ", cb " << callback_fn << ", cbdata " <<
 237            data);
 238
 239 #if STORE_CLIENT_LIST_DEBUG
 240
 241     assert(this == storeClientListSearch(entry->mem_obj, data));
 242 #endif
 243
 244     assert(!_callback.pending());
 245 #if ONLYCONTIGUOUSREQUESTS
 246
 247     assert(cmp_offset == copyRequest.offset);
 248 #endif
 249     /* range requests will skip into the body */
 250     cmp_offset = copyRequest.offset;
 251     _callback = Callback (callback_fn, cbdataReference(data));
 252     copyInto.data = copyRequest.data;
 253     copyInto.length = copyRequest.length;
 254     copyInto.offset = copyRequest.offset;
 255
 256     static bool copying (false);
 257     assert (!copying);
 258     copying = true;
 259     PROF_start(storeClient_kickReads);
 260     /* we might be blocking comm reads due to readahead limits
 261      * now we have a new offset, trigger those reads...
 262      */
 263     entry->mem_obj->kickReads();
 264     PROF_stop(storeClient_kickReads);
 265     copying = false;
 266
 267     anEntry->lock("store_client::copy"); // see deletion note below
 268
 269     storeClientCopy2(entry, this);
 270
 271     // Bug 3480: This store_client object may be deleted now if, for example,
 272     // the client rejects the hit response copied above. Use on-stack pointers!
 273
 274 #if USE_ADAPTATION
 275     anEntry->kickProducer();
 276 #endif
 277     anEntry->unlock("store_client::copy");
 278 }
 279
 280 /*
 281  * This function is used below to decide if we have any more data to
 282  * send to the client.  If the store_status is STORE_PENDING, then we
 283  * do have more data to send.  If its STORE_OK, then
 284  * we continue checking.  If the object length is negative, then we
 285  * don't know the real length and must open the swap file to find out.
 286  * If the length is >= 0, then we compare it to the requested copy
 287  * offset.
 288  */
 289 static int
 290 storeClientNoMoreToSend(StoreEntry * e, store_client * sc)
 291 {
 292     int64_t len;
 293
 294     if (e->store_status == STORE_PENDING)
 295         return 0;
 296
 297     if ((len = e->objectLen()) < 0)
 298         return 0;
 299
 300     if (sc->copyInto.offset < len)
 301         return 0;
 302
 303     return 1;
 304 }
 305
 306 static void
 307 storeClientCopy2(StoreEntry * e, store_client * sc)
 308 {
 309     /* reentrancy not allowed  - note this could lead to
 310      * dropped events
 311      */
 312
 313     if (sc->flags.copy_event_pending) {
 314         return;
 315     }
 316
 317     if (EBIT_TEST(e->flags, ENTRY_FWD_HDR_WAIT)) {
 318         debugs(90, 5, "storeClientCopy2: returning because ENTRY_FWD_HDR_WAIT set");
 319         return;
 320     }
 321
 322     if (sc->flags.store_copying) {
 323         sc->flags.copy_event_pending = true;
 324         debugs(90, 3, "storeClientCopy2: Queueing storeClientCopyEvent()");
 325         eventAdd("storeClientCopyEvent", storeClientCopyEvent, sc, 0.0, 0);
 326         return;
 327     }
 328
 329     debugs(90, 3, "storeClientCopy2: " << e->getMD5Text());
 330     assert(sc->_callback.pending());
 331     /*
 332      * We used to check for ENTRY_ABORTED here.  But there were some
 333      * problems.  For example, we might have a slow client (or two) and
 334      * the server-side is reading far ahead and swapping to disk.  Even
 335      * if the server-side aborts, we want to give the client(s)
 336      * everything we got before the abort condition occurred.
 337      */
 338     /* Warning: doCopy may indirectly free itself in callbacks,
 339      * hence the lock to keep it active for the duration of
 340      * this function
 341      */
 342     cbdataInternalLock(sc);
 343     assert (!sc->flags.store_copying);
 344     sc->doCopy(e);
 345     assert (!sc->flags.store_copying);
 346     cbdataInternalUnlock(sc);
 347 }
 348
 349 void
 350 store_client::doCopy(StoreEntry *anEntry)
 351 {
 352     assert (anEntry == entry);
 353     flags.store_copying = true;
 354     MemObject *mem = entry->mem_obj;
 355
 356     debugs(33, 5, "store_client::doCopy: co: " <<
 357            copyInto.offset << ", hi: " <<
 358            mem->endOffset());
 359
 360     if (storeClientNoMoreToSend(entry, this)) {
 361         /* There is no more to send! */
 362         debugs(33, 3, HERE << "There is no more to send!");
 363         callback(0);
 364         flags.store_copying = false;
 365         return;
 366     }
 367
 368     /* Check that we actually have data */
 369     if (anEntry->store_status == STORE_PENDING && copyInto.offset >= mem->endOffset()) {
 370         debugs(90, 3, "store_client::doCopy: Waiting for more");
 371         flags.store_copying = false;
 372         return;
 373     }
 374
 375     /*
 376      * Slight weirdness here.  We open a swapin file for any
 377      * STORE_DISK_CLIENT, even if we can copy the requested chunk
 378      * from memory in the next block.  We must try to open the
 379      * swapin file before sending any data to the client side.  If
 380      * we postpone the open, and then can not open the file later
 381      * on, the client loses big time.  Its transfer just gets cut
 382      * off.  Better to open it early (while the client side handler
 383      * is clientCacheHit) so that we can fall back to a cache miss
 384      * if needed.
 385      */
 386
 387     if (STORE_DISK_CLIENT == getType() && swapin_sio == NULL)
 388         startSwapin();
 389     else
 390         scheduleRead();
 391 }
 392
 393 void
 394 store_client::startSwapin()
 395 {
 396     debugs(90, 3, "store_client::doCopy: Need to open swap in file");
 397     /* gotta open the swapin file */
 398
 399     if (storeTooManyDiskFilesOpen()) {
 400         /* yuck -- this causes a TCP_SWAPFAIL_MISS on the client side */
 401         fail();
 402         flags.store_copying = false;
 403         return;
 404     } else if (!flags.disk_io_pending) {
 405         /* Don't set store_io_pending here */
 406         storeSwapInStart(this);
 407
 408         if (swapin_sio == NULL) {
 409             fail();
 410             flags.store_copying = false;
 411             return;
 412         }
 413
 414         /*
 415          * If the open succeeds we either copy from memory, or
 416          * schedule a disk read in the next block.
 417          */
 418         scheduleRead();
 419
 420         return;
 421     } else {
 422         debugs(90, DBG_IMPORTANT, "WARNING: Averted multiple fd operation (1)");
 423         flags.store_copying = false;
 424         return;
 425     }
 426 }
 427
 428 void
 429 store_client::scheduleRead()
 430 {
 431     MemObject *mem = entry->mem_obj;
 432
 433     if (copyInto.offset >= mem->inmem_lo && copyInto.offset < mem->endOffset())
 434         scheduleMemRead();
 435     else
 436         scheduleDiskRead();
 437 }
 438
 439 void
 440 store_client::scheduleDiskRead()
 441 {
 442     /* What the client wants is not in memory. Schedule a disk read */
 443     assert(STORE_DISK_CLIENT == getType());
 444
 445     assert(!flags.disk_io_pending);
 446
 447     debugs(90, 3, "store_client::doCopy: reading from STORE");
 448
 449     fileRead();
 450
 451     flags.store_copying = false;
 452 }
 453
 454 void
 455 store_client::scheduleMemRead()
 456 {
 457     /* What the client wants is in memory */
 458     /* Old style */
 459     debugs(90, 3, "store_client::doCopy: Copying normal from memory");
 460     size_t sz = entry->mem_obj->data_hdr.copy(copyInto);
 461     callback(sz);
 462     flags.store_copying = false;
 463 }
 464
 465 void
 466 store_client::fileRead()
 467 {
 468     MemObject *mem = entry->mem_obj;
 469
 470     assert(_callback.pending());
 471     assert(!flags.disk_io_pending);
 472     flags.disk_io_pending = true;
 473
 474     if (mem->swap_hdr_sz != 0)
 475         if (entry->swap_status == SWAPOUT_WRITING)
 476             assert(mem->swapout.sio->offset() > copyInto.offset + (int64_t)mem->swap_hdr_sz);
 477
 478     storeRead(swapin_sio,
 479               copyInto.data,
 480               copyInto.length,
 481               copyInto.offset + mem->swap_hdr_sz,
 482               mem->swap_hdr_sz == 0 ? storeClientReadHeader
 483               : storeClientReadBody,
 484               this);
 485 }
 486
 487 void
 488 store_client::readBody(const char *buf, ssize_t len)
 489 {
 490     int parsed_header = 0;
 491
 492     // Don't assert disk_io_pending here.. may be called by read_header
 493     flags.disk_io_pending = false;
 494     assert(_callback.pending());
 495     debugs(90, 3, "storeClientReadBody: len " << len << "");
 496
 497     if (copyInto.offset == 0 && len > 0 && entry->getReply()->sline.status() == Http::scNone) {
 498         /* Our structure ! */
 499         HttpReply *rep = (HttpReply *) entry->getReply(); // bypass const
 500
 501         if (!rep->parseCharBuf(copyInto.data, headersEnd(copyInto.data, len))) {
 502             debugs(90, DBG_CRITICAL, "Could not parse headers from on disk object");
 503         } else {
 504             parsed_header = 1;
 505         }
 506     }
 507
 508     const HttpReply *rep = entry->getReply();
 509     if (len > 0 && rep && entry->mem_obj->inmem_lo == 0 && entry->objectLen() <= (int64_t)Config.Store.maxInMemObjSize && Config.onoff.memory_cache_disk) {
 510         storeGetMemSpace(len);
 511         // The above may start to free our object so we need to check again
 512         if (entry->mem_obj->inmem_lo == 0) {
 513             /* Copy read data back into memory.
 514              * copyInto.offset includes headers, which is what mem cache needs
 515              */
 516             int64_t mem_offset = entry->mem_obj->endOffset();
 517             if ((copyInto.offset == mem_offset) || (parsed_header && mem_offset == rep->hdr_sz)) {
 518                 entry->mem_obj->write(StoreIOBuffer(len, copyInto.offset, copyInto.data));
 519             }
 520         }
 521     }
 522
 523     callback(len);
 524 }
 525
 526 void
 527 store_client::fail()
 528 {
 529     object_ok = false;
 530     /* synchronous open failures callback from the store,
 531      * before startSwapin detects the failure.
 532      * TODO: fix this inconsistent behaviour - probably by
 533      * having storeSwapInStart become a callback functions,
 534      * not synchronous
 535      */
 536
 537     if (_callback.pending())
 538         callback(0, true);
 539 }
 540
 541 static void
 542 storeClientReadHeader(void *data, const char *buf, ssize_t len, StoreIOState::Pointer self)
 543 {
 544     store_client *sc = (store_client *)data;
 545     sc->readHeader(buf, len);
 546 }
 547
 548 static void
 549 storeClientReadBody(void *data, const char *buf, ssize_t len, StoreIOState::Pointer self)
 550 {
 551     store_client *sc = (store_client *)data;
 552     sc->readBody(buf, len);
 553 }
 554
 555 void
 556 store_client::unpackHeader(char const *buf, ssize_t len)
 557 {
 558     debugs(90, 3, "store_client::unpackHeader: len " << len << "");
 559
 560     if (len < 0) {
 561         debugs(90, 3, "store_client::unpackHeader: " << xstrerror() << "");
 562         fail();
 563         return;
 564     }
 565
 566     int swap_hdr_sz = 0;
 567     StoreMetaUnpacker aBuilder(buf, len, &swap_hdr_sz);
 568
 569     if (!aBuilder.isBufferSane()) {
 570         /* oops, bad disk file? */
 571         debugs(90, DBG_IMPORTANT, "WARNING: swapfile header inconsistent with available data");
 572         fail();
 573         return;
 574     }
 575
 576     tlv *tlv_list = aBuilder.createStoreMeta ();
 577
 578     if (tlv_list == NULL) {
 579         debugs(90, DBG_IMPORTANT, "WARNING: failed to unpack meta data");
 580         fail();
 581         return;
 582     }
 583
 584     /*
 585      * Check the meta data and make sure we got the right object.
 586      */
 587     for (tlv *t = tlv_list; t; t = t->next) {
 588         if (!t->checkConsistency(entry)) {
 589             storeSwapTLVFree(tlv_list);
 590             fail();
 591             return;
 592         }
 593     }
 594
 595     storeSwapTLVFree(tlv_list);
 596
 597     assert(swap_hdr_sz >= 0);
 598     assert(entry->swap_file_sz > 0);
 599     assert(entry->swap_file_sz >= static_cast<uint64_t>(swap_hdr_sz));
 600     entry->mem_obj->swap_hdr_sz = swap_hdr_sz;
 601     entry->mem_obj->object_sz = entry->swap_file_sz - swap_hdr_sz;
 602     debugs(90, 5, "store_client::unpackHeader: swap_file_sz=" <<
 603            entry->swap_file_sz << "( " << swap_hdr_sz << " + " <<
 604            entry->mem_obj->object_sz << ")");
 605 }
 606
 607 void
 608 store_client::readHeader(char const *buf, ssize_t len)
 609 {
 610     MemObject *const mem = entry->mem_obj;
 611
 612     assert(flags.disk_io_pending);
 613     flags.disk_io_pending = false;
 614     assert(_callback.pending());
 615
 616     unpackHeader (buf, len);
 617
 618     if (!object_ok)
 619         return;
 620
 621     /*
 622      * If our last read got some data the client wants, then give
 623      * it to them, otherwise schedule another read.
 624      */
 625     size_t body_sz = len - mem->swap_hdr_sz;
 626
 627     if (copyInto.offset < static_cast<int64_t>(body_sz)) {
 628         /*
 629          * we have (part of) what they want
 630          */
 631         size_t copy_sz = min(copyInto.length, body_sz);
 632         debugs(90, 3, "storeClientReadHeader: copying " << copy_sz << " bytes of body");
 633         memmove(copyInto.data, copyInto.data + mem->swap_hdr_sz, copy_sz);
 634
 635         readBody(copyInto.data, copy_sz);
 636
 637         return;
 638     }
 639
 640     /*
 641      * we don't have what the client wants, but at least we now
 642      * know the swap header size.
 643      */
 644     fileRead();
 645 }
 646
 647 int
 648 storeClientCopyPending(store_client * sc, StoreEntry * e, void *data)
 649 {
 650 #if STORE_CLIENT_LIST_DEBUG
 651     assert(sc == storeClientListSearch(e->mem_obj, data));
 652 #endif
 653 #ifndef SILLY_CODE
 654
 655     assert(sc);
 656 #endif
 657
 658     assert(sc->entry == e);
 659 #if SILLY_CODE
 660
 661     if (sc == NULL)
 662         return 0;
 663
 664 #endif
 665
 666     if (!sc->_callback.pending())
 667         return 0;
 668
 669     return 1;
 670 }
 671
 672 /*
 673  * This routine hasn't been optimised to take advantage of the
 674  * passed sc. Yet.
 675  */
 676 int
 677 storeUnregister(store_client * sc, StoreEntry * e, void *data)
 678 {
 679     MemObject *mem = e->mem_obj;
 680 #if STORE_CLIENT_LIST_DEBUG
 681
 682     assert(sc == storeClientListSearch(e->mem_obj, data));
 683 #endif
 684
 685     if (mem == NULL)
 686         return 0;
 687
 688     debugs(90, 3, "storeUnregister: called for '" << e->getMD5Text() << "'");
 689
 690     if (sc == NULL) {
 691         debugs(90, 3, "storeUnregister: No matching client for '" << e->getMD5Text() << "'");
 692         return 0;
 693     }
 694
 695     if (mem->clientCount() == 0) {
 696         debugs(90, 3, "storeUnregister: Consistency failure - store client being unregistered is not in the mem object's list for '" << e->getMD5Text() << "'");
 697         return 0;
 698     }
 699
 700     dlinkDelete(&sc->node, &mem->clients);
 701     -- mem->nclients;
 702
 703     if (e->store_status == STORE_OK && e->swap_status != SWAPOUT_DONE)
 704         e->swapOut();
 705
 706     if (sc->swapin_sio != NULL) {
 707         storeClose(sc->swapin_sio, StoreIOState::readerDone);
 708         sc->swapin_sio = NULL;
 709         ++statCounter.swap.ins;
 710     }
 711
 712     if (sc->_callback.pending()) {
 713         /* callback with ssize = -1 to indicate unexpected termination */
 714         debugs(90, 3, "storeUnregister: store_client for " << mem->url << " has a callback");
 715         sc->fail();
 716     }
 717
 718 #if STORE_CLIENT_LIST_DEBUG
 719     cbdataReferenceDone(sc->owner);
 720
 721 #endif
 722
 723     delete sc;
 724
 725     assert(e->lock_count > 0);
 726
 727     if (mem->nclients == 0)
 728         CheckQuickAbort(e);
 729     else
 730         mem->kickReads();
 731
 732 #if USE_ADAPTATION
 733     e->kickProducer();
 734 #endif
 735
 736     return 1;
 737 }
 738
 739 /* Call handlers waiting for  data to be appended to E. */
 740 void
 741 StoreEntry::invokeHandlers()
 742 {
 743     /* Commit what we can to disk, if appropriate */
 744     swapOut();
 745     int i = 0;
 746     store_client *sc;
 747     dlink_node *nx = NULL;
 748     dlink_node *node;
 749
 750     PROF_start(InvokeHandlers);
 751
 752     debugs(90, 3, "InvokeHandlers: " << getMD5Text()  );
 753     /* walk the entire list looking for valid callbacks */
 754
 755     for (node = mem_obj->clients.head; node; node = nx) {
 756         sc = (store_client *)node->data;
 757         nx = node->next;
 758         debugs(90, 3, "StoreEntry::InvokeHandlers: checking client #" << i  );
 759         ++i;
 760
 761         if (!sc->_callback.pending())
 762             continue;
 763
 764         if (sc->flags.disk_io_pending)
 765             continue;
 766
 767         storeClientCopy2(this, sc);
 768     }
 769     PROF_stop(InvokeHandlers);
 770 }
 771
 772 // XXX: Does not account for remote readers of local writers, causing
 773 // premature StoreEntry aborts.
 774 int
 775 storePendingNClients(const StoreEntry * e)
 776 {
 777     MemObject *mem = e->mem_obj;
 778     int npend = NULL == mem ? 0 : mem->nclients;
 779     debugs(90, 3, "storePendingNClients: returning " << npend);
 780     return npend;
 781 }
 782
 783 /* return true if the request should be aborted */
 784 static bool
 785 CheckQuickAbortIsReasonable(StoreEntry * entry)
 786 {
 787     MemObject * const mem = entry->mem_obj;
 788     assert(mem);
 789     debugs(90, 3, "entry=" << entry << ", mem=" << mem);
 790
 791     if (mem->request && !mem->request->flags.cachable) {
 792         debugs(90, 3, "quick-abort? YES !mem->request->flags.cachable");
 793         return true;
 794     }
 795
 796     if (EBIT_TEST(entry->flags, KEY_PRIVATE)) {
 797         debugs(90, 3, "quick-abort? YES KEY_PRIVATE");
 798         return true;
 799     }
 800
 801     int64_t expectlen = entry->getReply()->content_length + entry->getReply()->hdr_sz;
 802
 803     if (expectlen < 0) {
 804         /* expectlen is < 0 if *no* information about the object has been received */
 805         debugs(90, 3, "quick-abort? YES no object data received yet");
 806         return true;
 807     }
 808
 809     int64_t curlen =  mem->endOffset();
 810
 811     if (Config.quickAbort.min < 0) {
 812         debugs(90, 3, "quick-abort? NO disabled");
 813         return false;
 814     }
 815
 816     if (mem->request && mem->request->range && mem->request->getRangeOffsetLimit() < 0) {
 817         /* Don't abort if the admin has configured range_ofset -1 to download fully for caching. */
 818         debugs(90, 3, "quick-abort? NO admin configured range replies to full-download");
 819         return false;
 820     }
 821
 822     if (curlen > expectlen) {
 823         debugs(90, 3, "quick-abort? YES bad content length");
 824         return true;
 825     }
 826
 827     if ((expectlen - curlen) < (Config.quickAbort.min << 10)) {
 828         debugs(90, 3, "quick-abort? NO only a little more object left to receive");
 829         return false;
 830     }
 831
 832     if ((expectlen - curlen) > (Config.quickAbort.max << 10)) {
 833         debugs(90, 3, "quick-abort? YES too much left to go");
 834         return true;
 835     }
 836
 837     if (expectlen < 100) {
 838         debugs(90, 3, "quick-abort? NO avoid FPE");
 839         return false;
 840     }
 841
 842     if ((curlen / (expectlen / 100)) > (Config.quickAbort.pct)) {
 843         debugs(90, 3, "quick-abort? NO past point of no return");
 844         return false;
 845     }
 846
 847     debugs(90, 3, "quick-abort? YES default");
 848     return true;
 849 }
 850
 851 static void
 852 CheckQuickAbort(StoreEntry * entry)
 853 {
 854     assert (entry);
 855
 856     if (storePendingNClients(entry) > 0)
 857         return;
 858
 859     if (entry->store_status != STORE_PENDING)
 860         return;
 861
 862     if (EBIT_TEST(entry->flags, ENTRY_SPECIAL))
 863         return;
 864
 865     if (!CheckQuickAbortIsReasonable(entry))
 866         return;
 867
 868     entry->abort();
 869 }
 870
 871 void
 872 store_client::dumpStats(MemBuf * output, int clientNumber) const
 873 {
 874     if (_callback.pending())
 875         return;
 876
 877     output->Printf("\tClient #%d, %p\n", clientNumber, _callback.callback_data);
 878
 879     output->Printf("\t\tcopy_offset: %" PRId64 "\n",
 880                    copyInto.offset);
 881
 882     output->Printf("\t\tcopy_size: %d\n",
 883                    (int) copyInto.length);
 884
 885     output->Printf("\t\tflags:");
 886
 887     if (flags.disk_io_pending)
 888         output->Printf(" disk_io_pending");
 889
 890     if (flags.store_copying)
 891         output->Printf(" store_copying");
 892
 893     if (flags.copy_event_pending)
 894         output->Printf(" copy_event_pending");
 895
 896     output->Printf("\n");
 897 }
 898
 899 bool
 900 store_client::Callback::pending() const
 901 {
 902     return callback_handler && callback_data;
 903 }
 904
 905 store_client::Callback::Callback(STCB *function, void *data) : callback_handler(function), callback_data (data) {}
 906
 907 #if USE_DELAY_POOLS
 908 void
 909 store_client::setDelayId(DelayId delay_id)
 910 {
 911     delayId = delay_id;
 912 }
 913 #endif