store_client.cc

Go to the documentation of this file.

/*
 * Copyright (C) 1996-2023 The Squid Software Foundation and contributors
 *
 * Squid software is distributed under GPLv2+ license and includes
 * contributions from numerous individuals and organizations.
 * Please see the COPYING and CONTRIBUTORS files for details.
 */
 
/* DEBUG: section 90    Storage Manager Client-Side Interface */
 
#include "squid.h"
#include "acl/FilledChecklist.h"
#include "base/AsyncCbdataCalls.h"
#include "base/CodeContext.h"
#include "event.h"
#include "globals.h"
#include "HttpReply.h"
#include "HttpRequest.h"
#include "MemBuf.h"
#include "MemObject.h"
#include "mime_header.h"
#include "sbuf/Stream.h"
#include "SquidConfig.h"
#include "SquidMath.h"
#include "StatCounters.h"
#include "Store.h"
#include "store/SwapMetaIn.h"
#include "store_swapin.h"
#include "StoreClient.h"
#if USE_DELAY_POOLS
#include "DelayPools.h"
#endif
 
/*
 * NOTE: 'Header' refers to the swapfile metadata header.
 *   'OBJHeader' refers to the object header, with canonical
 *   processed object headers (which may derive from FTP/HTTP etc
 *   upstream protocols
 *       'Body' refers to the swapfile body, which is the full
 *        HTTP reply (including HTTP headers and body).
 */
static StoreIOState::STRCB storeClientReadBody;
static StoreIOState::STRCB storeClientReadHeader;
static void storeClientCopy2(StoreEntry * e, store_client * sc);
static bool CheckQuickAbortIsReasonable(StoreEntry * entry);
 
CBDATA_CLASS_INIT(store_client);
 
/* StoreClient */
 
bool
StoreClient::onCollapsingPath() const
{
    if (!Config.onoff.collapsed_forwarding)
        return false;
 
    if (!Config.accessList.collapsedForwardingAccess)
        return true;
 
    ACLFilledChecklist checklist(Config.accessList.collapsedForwardingAccess, nullptr);
    fillChecklist(checklist);
    return checklist.fastCheck().allowed();
}
 
bool
StoreClient::startCollapsingOn(const StoreEntry &e, const bool doingRevalidation) const
{
    if (!e.hittingRequiresCollapsing())
        return false; // collapsing is impossible due to the entry state
 
    if (!onCollapsingPath())
        return false; // collapsing is impossible due to Squid configuration
 
    /* collapsing is possible; the caller must collapse */
 
    if (const auto tags = loggingTags()) {
        if (doingRevalidation)
            tags->collapsingHistory.revalidationCollapses++;
        else
            tags->collapsingHistory.otherCollapses++;
    }
 
    didCollapse = true;
    debugs(85, 5, e << " doingRevalidation=" << doingRevalidation);
    return true;
}
 
/* store_client */
 
int
store_client::getType() const
{
    return type;
}
 
#if STORE_CLIENT_LIST_DEBUG
static store_client *
storeClientListSearch(const MemObject * mem, void *data)
{
    dlink_node *node;
    store_client *sc = nullptr;
 
    for (node = mem->clients.head; node; node = node->next) {
        sc = node->data;
 
        if (sc->owner == data)
            return sc;
    }
 
    return nullptr;
}
 
int
storeClientIsThisAClient(store_client * sc, void *someClient)
{
    return sc->owner == someClient;
}
 
#endif
#include "HttpRequest.h"
 
/* add client with fd to client list */
store_client *
storeClientListAdd(StoreEntry * e, void *data)
{
    MemObject *mem = e->mem_obj;
    store_client *sc;
    assert(mem);
#if STORE_CLIENT_LIST_DEBUG
 
    if (storeClientListSearch(mem, data) != NULL)
        /* XXX die! */
        assert(1 == 0);
#else
    (void)data;
#endif
 
    sc = new store_client (e);
 
    mem->addClient(sc);
 
    return sc;
}
 
void
store_client::FinishCallback(store_client * const sc)
{
    sc->finishCallback();
}
 
void
store_client::finishCallback()
{
    Assure(_callback.callback_handler);
    Assure(_callback.notifier);
 
    // XXX: Some legacy code relies on zero-length buffers having nil data
    // pointers. Some other legacy code expects "correct" result.offset even
    // when there is no body to return. Accommodate all those expectations.
    auto result = StoreIOBuffer(0, copyInto.offset, nullptr);
    if (object_ok && parsingBuffer && parsingBuffer->contentSize())
        result = parsingBuffer->packBack();
    result.flags.error = object_ok ? 0 : 1;
 
    // TODO: Move object_ok handling above into this `if` statement.
    if (object_ok) {
        // works for zero hdr_sz cases as well; see also: nextHttpReadOffset()
        discardableHttpEnd_ = NaturalSum<int64_t>(entry->mem().baseReply().hdr_sz, result.offset, result.length).value();
    } else {
        // object_ok is sticky, so we will not be able to use any response bytes
        discardableHttpEnd_ = entry->mem().endOffset();
    }
    debugs(90, 7, "with " << result << "; discardableHttpEnd_=" << discardableHttpEnd_);
 
    // no HTTP headers and no body bytes (but not because there was no space)
    atEof_ = !sendingHttpHeaders() && !result.length && copyInto.length;
 
    parsingBuffer.reset();
    ++answers;
 
    STCB *temphandler = _callback.callback_handler;
    const auto cbdata = _callback.cbData.validDone();
    _callback = Callback();
    copyInto.data = nullptr;
 
    if (cbdata)
        temphandler(cbdata, result);
}
 
store_client::store_client(StoreEntry *e) :
#if STORE_CLIENT_LIST_DEBUG
    owner(cbdataReference(data)),
#endif
    entry(e),
    type(e->storeClientType()),
    object_ok(true),
    atEof_(false),
    answers(0)
{
    Assure(entry);
    entry->lock("store_client");
 
    flags.disk_io_pending = false;
    flags.store_copying = false;
    ++ entry->refcount;
 
    if (getType() == STORE_DISK_CLIENT) {
        /* assert we'll be able to get the data we want */
        /* maybe we should open swapin_sio here */
        assert(entry->hasDisk() && !entry->swapoutFailed());
    }
}
 
store_client::~store_client()
{
    assert(entry);
    entry->unlock("store_client");
}
 
/* copy bytes requested by the client */
void
storeClientCopy(store_client * sc,
                StoreEntry * e,
                StoreIOBuffer copyInto,
                STCB * callback,
                void *data)
{
    assert (sc != nullptr);
    sc->copy(e, copyInto,callback,data);
}
 
void
store_client::copy(StoreEntry * anEntry,
                   StoreIOBuffer copyRequest,
                   STCB * callback_fn,
                   void *data)
{
    assert (anEntry == entry);
    assert (callback_fn);
    assert (data);
    assert(!EBIT_TEST(entry->flags, ENTRY_ABORTED));
    debugs(90, 3, "store_client::copy: " << entry->getMD5Text() << ", from " <<
           copyRequest.offset << ", for length " <<
           (int) copyRequest.length << ", cb " << callback_fn << ", cbdata " <<
           data);
 
#if STORE_CLIENT_LIST_DEBUG
 
    assert(this == storeClientListSearch(entry->mem_obj, data));
#endif
 
    assert(!_callback.pending());
    _callback = Callback(callback_fn, data);
    copyInto.data = copyRequest.data;
    copyInto.length = copyRequest.length;
    copyInto.offset = copyRequest.offset;
    Assure(copyInto.offset >= 0);
 
    if (!copyInto.length) {
        // During the first storeClientCopy() call, a zero-size buffer means
        // that we will have to drop any HTTP response body bytes we read (with
        // the HTTP headers from disk). After that, it means we cannot return
        // anything to the caller at all.
        debugs(90, 2, "WARNING: zero-size storeClientCopy() buffer: " << copyInto);
        // keep going; moreToRead() should prevent any from-Store reading
    }
 
    // Our nextHttpReadOffset() expects the first copy() call to have zero
    // offset. More complex code could handle a positive first offset, but it
    // would only be useful when reading responses from memory: We would not
    // _delay_ the response (to read the requested HTTP body bytes from disk)
    // when we already can respond with HTTP headers.
    Assure(!copyInto.offset || answeredOnce());
 
    parsingBuffer.emplace(copyInto);
 
    discardableHttpEnd_ = nextHttpReadOffset();
    debugs(90, 7, "discardableHttpEnd_=" << discardableHttpEnd_);
 
    static bool copying (false);
    assert (!copying);
    copying = true;
    /* we might be blocking comm reads due to readahead limits
     * now we have a new offset, trigger those reads...
     */
    entry->mem_obj->kickReads();
    copying = false;
 
    anEntry->lock("store_client::copy"); // see deletion note below
 
    storeClientCopy2(entry, this);
 
    // Bug 3480: This store_client object may be deleted now if, for example,
    // the client rejects the hit response copied above. Use on-stack pointers!
 
#if USE_ADAPTATION
    anEntry->kickProducer();
#endif
    anEntry->unlock("store_client::copy");
 
    // Add no code here. This object may no longer exist.
}
 
bool
store_client::moreToRead() const
{
    if (!copyInto.length)
        return false; // the client supplied a zero-size buffer
 
    if (entry->store_status == STORE_PENDING)
        return true; // there may be more coming
 
    /* STORE_OK, including aborted entries: no more data is coming */
 
    if (canReadFromMemory())
        return true; // memory has the first byte wanted by the client
 
    if (!entry->hasDisk())
        return false; // cannot read anything from disk either
 
    if (entry->objectLen() >= 0 && copyInto.offset >= entry->contentLen())
        return false; // the disk cannot have byte(s) wanted by the client
 
    // we cannot be sure until we swap in metadata and learn contentLen(),
    // but the disk may have the byte(s) wanted by the client
    return true;
}
 
static void
storeClientCopy2(StoreEntry * e, store_client * sc)
{
    /* reentrancy not allowed  - note this could lead to
     * dropped notifications about response data availability
     */
 
    if (sc->flags.store_copying) {
        debugs(90, 3, "prevented recursive copying for " << *e);
        return;
    }
 
    debugs(90, 3, "storeClientCopy2: " << e->getMD5Text());
    assert(sc->_callback.pending());
    /*
     * We used to check for ENTRY_ABORTED here.  But there were some
     * problems.  For example, we might have a slow client (or two) and
     * the peer server is reading far ahead and swapping to disk.  Even
     * if the peer aborts, we want to give the client(s)
     * everything we got before the abort condition occurred.
     */
    sc->doCopy(e);
}
 
bool
store_client::sendingHttpHeaders() const
{
    return !answeredOnce() && entry->mem().baseReply().hdr_sz > 0;
}
 
void
store_client::doCopy(StoreEntry *anEntry)
{
    Assure(_callback.pending());
    Assure(!flags.disk_io_pending);
    Assure(!flags.store_copying);
 
    assert (anEntry == entry);
    flags.store_copying = true;
    MemObject *mem = entry->mem_obj;
 
    debugs(33, 5, this << " into " << copyInto <<
           " hi: " << mem->endOffset() <<
           " objectLen: " << entry->objectLen() <<
           " past_answers: " << answers);
 
    const auto sendHttpHeaders = sendingHttpHeaders();
 
    if (!sendHttpHeaders && !moreToRead()) {
        /* There is no more to send! */
        debugs(33, 3, "There is no more to send!");
        noteNews();
        flags.store_copying = false;
        return;
    }
 
    if (!sendHttpHeaders && anEntry->store_status == STORE_PENDING && nextHttpReadOffset() >= mem->endOffset()) {
        debugs(90, 3, "store_client::doCopy: Waiting for more");
        flags.store_copying = false;
        return;
    }
 
    /*
     * Slight weirdness here.  We open a swapin file for any
     * STORE_DISK_CLIENT, even if we can copy the requested chunk
     * from memory in the next block.  We must try to open the
     * swapin file before sending any data to the client side.  If
     * we postpone the open, and then can not open the file later
     * on, the client loses big time.  Its transfer just gets cut
     * off.  Better to open it early (while the client side handler
     * is clientCacheHit) so that we can fall back to a cache miss
     * if needed.
     */
 
    if (STORE_DISK_CLIENT == getType() && swapin_sio == nullptr) {
        if (!startSwapin())
            return; // failure
    }
 
    // Send any immediately available body bytes unless we sendHttpHeaders.
    // TODO: Send those body bytes when we sendHttpHeaders as well.
    if (!sendHttpHeaders && canReadFromMemory()) {
        readFromMemory();
        noteNews(); // will sendHttpHeaders (if needed) as well
        flags.store_copying = false;
        return;
    }
 
    if (sendHttpHeaders) {
        debugs(33, 5, "just send HTTP headers: " << mem->baseReply().hdr_sz);
        noteNews();
        flags.store_copying = false;
        return;
    }
 
    // no information that the client needs is available immediately
    scheduleDiskRead();
}
 
bool
store_client::startSwapin()
{
    debugs(90, 3, "store_client::doCopy: Need to open swap in file");
    /* gotta open the swapin file */
 
    if (storeTooManyDiskFilesOpen()) {
        /* yuck -- this causes a TCP_SWAPFAIL_MISS on the client side */
        fail();
        flags.store_copying = false;
        return false;
    } else if (!flags.disk_io_pending) {
        /* Don't set store_io_pending here */
        storeSwapInStart(this);
 
        if (swapin_sio == nullptr) {
            fail();
            flags.store_copying = false;
            return false;
        }
 
        return true;
    } else {
        debugs(90, DBG_IMPORTANT, "WARNING: Averted multiple fd operation (1)");
        flags.store_copying = false;
        return false;
    }
}
 
void
store_client::noteSwapInDone(const bool error)
{
    Assure(_callback.pending());
    if (error)
        fail();
    else
        noteNews();
}
 
void
store_client::scheduleDiskRead()
{
    /* What the client wants is not in memory. Schedule a disk read */
    if (getType() == STORE_DISK_CLIENT) {
        // we should have called startSwapin() already
        assert(swapin_sio != nullptr);
    } else if (!swapin_sio && !startSwapin()) {
        debugs(90, 3, "bailing after swapin start failure for " << *entry);
        assert(!flags.store_copying);
        return;
    }
 
    assert(!flags.disk_io_pending);
 
    debugs(90, 3, "reading " << *entry << " from disk");
 
    fileRead();
 
    flags.store_copying = false;
}
 
bool
store_client::canReadFromMemory() const
{
    const auto &mem = entry->mem();
    const auto memReadOffset = nextHttpReadOffset();
    // XXX: This (lo <= offset < end) logic does not support Content-Range gaps.
    return mem.inmem_lo <= memReadOffset && memReadOffset < mem.endOffset() &&
           parsingBuffer->spaceSize();
}
 
int64_t
store_client::nextHttpReadOffset() const
{
    Assure(parsingBuffer);
    const auto &mem = entry->mem();
    const auto hdr_sz = mem.baseReply().hdr_sz;
    // Certain SMP cache manager transactions do not store HTTP headers in
    // mem_hdr; they store just a kid-specific piece of the future report body.
    // In such cases, hdr_sz ought to be zero. In all other (known) cases,
    // mem_hdr contains HTTP response headers (positive hdr_sz if parsed)
    // followed by HTTP response body. This code math accommodates all cases.
    return NaturalSum<int64_t>(hdr_sz, copyInto.offset, parsingBuffer->contentSize()).value();
}
 
void
store_client::readFromMemory()
{
    Assure(parsingBuffer);
    const auto readInto = parsingBuffer->space().positionAt(nextHttpReadOffset());
 
    debugs(90, 3, "copying HTTP body bytes from memory into " << readInto);
    const auto sz = entry->mem_obj->data_hdr.copy(readInto);
    Assure(sz > 0); // our canReadFromMemory() precondition guarantees that
    parsingBuffer->appended(readInto.data, sz);
}
 
void
store_client::fileRead()
{
    MemObject *mem = entry->mem_obj;
 
    assert(_callback.pending());
    assert(!flags.disk_io_pending);
    flags.disk_io_pending = true;
 
    // mem->swap_hdr_sz is zero here during initial read(s)
    const auto nextStoreReadOffset = NaturalSum<int64_t>(mem->swap_hdr_sz, nextHttpReadOffset()).value();
 
    // XXX: If fileRead() is called when we do not yet know mem->swap_hdr_sz,
    // then we must start reading from disk offset zero to learn it: we cannot
    // compute correct HTTP response start offset on disk without it. However,
    // late startSwapin() calls imply that the assertion below might fail.
    Assure(mem->swap_hdr_sz > 0 || !nextStoreReadOffset);
 
    // TODO: Remove this assertion. Introduced in 1998 commit 3157c72, it
    // assumes that swapped out memory is freed unconditionally, but we no
    // longer do that because trimMemory() path checks lowestMemReaderOffset().
    // It is also misplaced: We are not swapping out anything here and should
    // not care about any swapout invariants.
    if (mem->swap_hdr_sz != 0)
        if (entry->swappingOut())
            assert(mem->swapout.sio->offset() > nextStoreReadOffset);
 
    // XXX: We should let individual cache_dirs limit the read size instead, but
    // we cannot do that without more fixes and research because:
    // * larger reads corrupt responses when cache_dir uses SharedMemory::get();
    // * we do not know how to find all I/O code that assumes this limit;
    // * performance effects of larger disk reads may be negative somewhere.
    const decltype(StoreIOBuffer::length) maxReadSize = SM_PAGE_SIZE;
 
    Assure(parsingBuffer);
    // also, do not read more than we can return (via a copyInto.length buffer)
    const auto readSize = std::min(copyInto.length, maxReadSize);
    lastDiskRead = parsingBuffer->makeSpace(readSize).positionAt(nextStoreReadOffset);
    debugs(90, 5, "into " << lastDiskRead);
 
    storeRead(swapin_sio,
              lastDiskRead.data,
              lastDiskRead.length,
              lastDiskRead.offset,
              mem->swap_hdr_sz == 0 ? storeClientReadHeader
              : storeClientReadBody,
              this);
}
 
void
store_client::readBody(const char * const buf, const ssize_t lastIoResult)
{
    Assure(flags.disk_io_pending);
    flags.disk_io_pending = false;
    assert(_callback.pending());
    Assure(parsingBuffer);
    debugs(90, 3, "got " << lastIoResult << " using " << *parsingBuffer);
 
    if (lastIoResult < 0)
        return fail();
 
    if (!lastIoResult) {
        if (answeredOnce())
            return noteNews();
 
        debugs(90, DBG_CRITICAL, "ERROR: Truncated HTTP headers in on-disk object");
        return fail();
    }
 
    assert(lastDiskRead.data == buf);
    lastDiskRead.length = lastIoResult;
 
    parsingBuffer->appended(buf, lastIoResult);
 
    // we know swap_hdr_sz by now and were reading beyond swap metadata because
    // readHead() would have been called otherwise (to read swap metadata)
    const auto swap_hdr_sz = entry->mem().swap_hdr_sz;
    Assure(swap_hdr_sz > 0);
    Assure(!Less(lastDiskRead.offset, swap_hdr_sz));
 
    // Map lastDiskRead (i.e. the disk area we just read) to an HTTP reply part.
    // The bytes are the same, but disk and HTTP offsets differ by swap_hdr_sz.
    const auto httpOffset = lastDiskRead.offset - swap_hdr_sz;
    const auto httpPart = StoreIOBuffer(lastDiskRead).positionAt(httpOffset);
 
    maybeWriteFromDiskToMemory(httpPart);
    handleBodyFromDisk();
}
 
void
store_client::handleBodyFromDisk()
{
    // We cannot de-serialize on-disk HTTP response without MemObject because
    // without MemObject::swap_hdr_sz we cannot know where that response starts.
    Assure(entry->mem_obj);
    Assure(entry->mem_obj->swap_hdr_sz > 0);
 
    if (!answeredOnce()) {
        // All on-disk responses have HTTP headers. First disk body read(s)
        // include HTTP headers that we must parse (if needed) and skip.
        const auto haveHttpHeaders = entry->hasParsedReplyHeader();
        if (!haveHttpHeaders && !parseHttpHeadersFromDisk())
            return;
        skipHttpHeadersFromDisk();
    }
 
    noteNews();
}
 
void
store_client::maybeWriteFromDiskToMemory(const StoreIOBuffer &httpResponsePart)
{
    // XXX: Reject [memory-]uncachable/unshareable responses instead of assuming
    // that an HTTP response should be written to MemObject's data_hdr (and that
    // it may purge already cached entries) just because it "fits" and was
    // loaded from disk. For example, this response may already be marked for
    // release. The (complex) cachability decision(s) should be made outside
    // (and obeyed by) this low-level code.
    if (httpResponsePart.length && entry->mem_obj->inmem_lo == 0 && entry->objectLen() <= (int64_t)Config.Store.maxInMemObjSize && Config.onoff.memory_cache_disk) {
        storeGetMemSpace(httpResponsePart.length);
        // XXX: This "recheck" is not needed because storeGetMemSpace() cannot
        // purge mem_hdr bytes of a locked entry, and we do lock ours. And
        // inmem_lo offset itself should not be relevant to appending new bytes.
        //
        // recheck for the above call may purge entry's data from the memory cache
        if (entry->mem_obj->inmem_lo == 0) {
            // XXX: This code assumes a non-shared memory cache.
            if (httpResponsePart.offset == entry->mem_obj->endOffset())
                entry->mem_obj->write(httpResponsePart);
        }
    }
}
 
void
store_client::fail()
{
    debugs(90, 3, (object_ok ? "once" : "again"));
    if (!object_ok)
        return; // we failed earlier; nothing to do now
 
    object_ok = false;
 
    noteNews();
}
 
void
store_client::noteNews()
{
    /* synchronous open failures callback from the store,
     * before startSwapin detects the failure.
     * TODO: fix this inconsistent behaviour - probably by
     * having storeSwapInStart become a callback functions,
     * not synchronous
     */
 
    if (!_callback.callback_handler) {
        debugs(90, 5, "client lost interest");
        return;
    }
 
    if (_callback.notifier) {
        debugs(90, 5, "earlier news is being delivered by " << _callback.notifier);
        return;
    }
 
    _callback.notifier = asyncCall(90, 4, "store_client::FinishCallback", cbdataDialer(store_client::FinishCallback, this));
    ScheduleCallHere(_callback.notifier);
 
    Assure(!_callback.pending());
}
 
static void
storeClientReadHeader(void *data, const char *buf, ssize_t len, StoreIOState::Pointer)
{
    store_client *sc = (store_client *)data;
    sc->readHeader(buf, len);
}
 
static void
storeClientReadBody(void *data, const char *buf, ssize_t len, StoreIOState::Pointer)
{
    store_client *sc = (store_client *)data;
    sc->readBody(buf, len);
}
 
void
store_client::readHeader(char const *buf, ssize_t len)
{
    MemObject *const mem = entry->mem_obj;
 
    assert(flags.disk_io_pending);
    flags.disk_io_pending = false;
    assert(_callback.pending());
 
    // abort if we fail()'d earlier
    if (!object_ok)
        return;
 
    Assure(parsingBuffer);
    debugs(90, 3, "got " << len << " using " << *parsingBuffer);
 
    if (len < 0)
        return fail();
 
    try {
        Assure(!parsingBuffer->contentSize());
        parsingBuffer->appended(buf, len);
        Store::UnpackHitSwapMeta(buf, len, *entry);
        parsingBuffer->consume(mem->swap_hdr_sz);
    } catch (...) {
        debugs(90, DBG_IMPORTANT, "ERROR: Failed to unpack Store entry metadata: " << CurrentException);
        fail();
        return;
    }
 
    maybeWriteFromDiskToMemory(parsingBuffer->content());
    handleBodyFromDisk();
}
 
/*
 * This routine hasn't been optimised to take advantage of the
 * passed sc. Yet.
 */
int
storeUnregister(store_client * sc, StoreEntry * e, void *data)
{
    MemObject *mem = e->mem_obj;
#if STORE_CLIENT_LIST_DEBUG
    assert(sc == storeClientListSearch(e->mem_obj, data));
#else
    (void)data;
#endif
 
    if (mem == nullptr)
        return 0;
 
    debugs(90, 3, "storeUnregister: called for '" << e->getMD5Text() << "'");
 
    if (sc == nullptr) {
        debugs(90, 3, "storeUnregister: No matching client for '" << e->getMD5Text() << "'");
        return 0;
    }
 
    if (mem->clientCount() == 0) {
        debugs(90, 3, "storeUnregister: Consistency failure - store client being unregistered is not in the mem object's list for '" << e->getMD5Text() << "'");
        return 0;
    }
 
    dlinkDelete(&sc->node, &mem->clients);
    -- mem->nclients;
 
    const auto swapoutFinished = e->swappedOut() || e->swapoutFailed();
    if (e->store_status == STORE_OK && !swapoutFinished)
        e->swapOut();
 
    if (sc->swapin_sio != nullptr) {
        storeClose(sc->swapin_sio, StoreIOState::readerDone);
        sc->swapin_sio = nullptr;
        ++statCounter.swap.ins;
    }
 
    if (sc->_callback.callback_handler || sc->_callback.notifier) {
        debugs(90, 3, "forgetting store_client callback for " << *e);
        // Do not notify: Callers want to stop copying and forget about this
        // pending copy request. Some would mishandle a notification from here.
        if (sc->_callback.notifier)
            sc->_callback.notifier->cancel("storeUnregister");
    }
 
#if STORE_CLIENT_LIST_DEBUG
    cbdataReferenceDone(sc->owner);
 
#endif
 
    // We must lock to safely dereference e below, after deleting sc and after
    // calling CheckQuickAbortIsReasonable().
    e->lock("storeUnregister");
 
    // XXX: We might be inside sc store_client method somewhere up the call
    // stack. TODO: Convert store_client to AsyncJob to make destruction async.
    delete sc;
 
    if (CheckQuickAbortIsReasonable(e))
        e->abort();
    else
        mem->kickReads();
 
#if USE_ADAPTATION
    e->kickProducer();
#endif
 
    e->unlock("storeUnregister");
    return 1;
}
 
/* Call handlers waiting for  data to be appended to E. */
void
StoreEntry::invokeHandlers()
{
    if (EBIT_TEST(flags, DELAY_SENDING)) {
        debugs(90, 3, "DELAY_SENDING is on, exiting " << *this);
        return;
    }
    if (EBIT_TEST(flags, ENTRY_FWD_HDR_WAIT)) {
        debugs(90, 3, "ENTRY_FWD_HDR_WAIT is on, exiting " << *this);
        return;
    }
 
    /* Commit what we can to disk, if appropriate */
    swapOut();
    int i = 0;
    store_client *sc;
    dlink_node *nx = nullptr;
    dlink_node *node;
 
    debugs(90, 3, mem_obj->nclients << " clients; " << *this << ' ' << getMD5Text());
    /* walk the entire list looking for valid callbacks */
 
    const auto savedContext = CodeContext::Current();
    for (node = mem_obj->clients.head; node; node = nx) {
        sc = (store_client *)node->data;
        nx = node->next;
        ++i;
 
        if (!sc->_callback.pending())
            continue;
 
        if (sc->flags.disk_io_pending)
            continue;
 
        if (sc->flags.store_copying)
            continue;
 
        // XXX: If invokeHandlers() is (indirectly) called from a store_client
        // method, then the above three conditions may not be sufficient to
        // prevent us from reentering the same store_client object! This
        // probably does not happen in the current code, but no observed
        // invariant prevents this from (accidentally) happening in the future.
 
        // TODO: Convert store_client into AsyncJob; make this call asynchronous
        CodeContext::Reset(sc->_callback.codeContext);
        debugs(90, 3, "checking client #" << i);
        storeClientCopy2(this, sc);
    }
    CodeContext::Reset(savedContext);
}
 
// Does not account for remote readers/clients.
int
storePendingNClients(const StoreEntry * e)
{
    MemObject *mem = e->mem_obj;
    int npend = nullptr == mem ? 0 : mem->nclients;
    debugs(90, 3, "storePendingNClients: returning " << npend);
    return npend;
}
 
/* return true if the request should be aborted */
static bool
CheckQuickAbortIsReasonable(StoreEntry * entry)
{
    assert(entry);
    debugs(90, 3, "entry=" << *entry);
 
    if (storePendingNClients(entry) > 0) {
        debugs(90, 3, "quick-abort? NO storePendingNClients() > 0");
        return false;
    }
 
    if (Store::Root().transientReaders(*entry)) {
        debugs(90, 3, "quick-abort? NO still have one or more transient readers");
        return false;
    }
 
    if (entry->store_status != STORE_PENDING) {
        debugs(90, 3, "quick-abort? NO store_status != STORE_PENDING");
        return false;
    }
 
    if (EBIT_TEST(entry->flags, ENTRY_SPECIAL)) {
        debugs(90, 3, "quick-abort? NO ENTRY_SPECIAL");
        return false;
    }
 
    if (shutting_down) {
        debugs(90, 3, "quick-abort? YES avoid heavy optional work during shutdown");
        return true;
    }
 
    MemObject * const mem = entry->mem_obj;
    assert(mem);
    debugs(90, 3, "mem=" << mem);
 
    if (mem->request && !mem->request->flags.cachable) {
        debugs(90, 3, "quick-abort? YES !mem->request->flags.cachable");
        return true;
    }
 
    if (EBIT_TEST(entry->flags, KEY_PRIVATE)) {
        debugs(90, 3, "quick-abort? YES KEY_PRIVATE");
        return true;
    }
 
    const auto &reply = mem->baseReply();
 
    if (reply.hdr_sz <= 0) {
        // TODO: Check whether this condition works for HTTP/0 responses.
        debugs(90, 3, "quick-abort? YES no object data received yet");
        return true;
    }
 
    if (Config.quickAbort.min < 0) {
        debugs(90, 3, "quick-abort? NO disabled");
        return false;
    }
 
    if (mem->request && mem->request->range && mem->request->getRangeOffsetLimit() < 0) {
        // the admin has configured "range_offset_limit none"
        debugs(90, 3, "quick-abort? NO admin configured range replies to full-download");
        return false;
    }
 
    if (reply.content_length < 0) {
        // XXX: cf.data.pre does not document what should happen in this case
        // We know that quick_abort is enabled, but no limit can be applied.
        debugs(90, 3, "quick-abort? YES unknown content length");
        return true;
    }
    const auto expectlen = reply.hdr_sz + reply.content_length;
 
    int64_t curlen =  mem->endOffset();
 
    if (curlen > expectlen) {
        debugs(90, 3, "quick-abort? YES bad content length (" << curlen << " of " << expectlen << " bytes received)");
        return true;
    }
 
    if ((expectlen - curlen) < (Config.quickAbort.min << 10)) {
        debugs(90, 3, "quick-abort? NO only a little more object left to receive");
        return false;
    }
 
    if ((expectlen - curlen) > (Config.quickAbort.max << 10)) {
        debugs(90, 3, "quick-abort? YES too much left to go");
        return true;
    }
 
    if (curlen > expectlen*(Config.quickAbort.pct/100.0)) {
        debugs(90, 3, "quick-abort? NO past point of no return");
        return false;
    }
 
    debugs(90, 3, "quick-abort? YES default");
    return true;
}
 
bool
store_client::parseHttpHeadersFromDisk()
{
    try {
        return tryParsingHttpHeaders();
    } catch (...) {
        // XXX: Our parser enforces Config.maxReplyHeaderSize limit, but our
        // packer does not. Since packing might increase header size, we may
        // cache a header that we cannot parse and get here. Same for MemStore.
        debugs(90, DBG_CRITICAL, "ERROR: Cannot parse on-disk HTTP headers" <<
               Debug::Extra << "exception: " << CurrentException <<
               Debug::Extra << "raw input size: " << parsingBuffer->contentSize() << " bytes" <<
               Debug::Extra << "current buffer capacity: " << parsingBuffer->capacity() << " bytes");
        fail();
        return false;
    }
}
 
bool
store_client::tryParsingHttpHeaders()
{
    Assure(parsingBuffer);
    Assure(!copyInto.offset); // otherwise, parsingBuffer cannot have HTTP response headers
    auto &adjustableReply = entry->mem().adjustableBaseReply();
    if (adjustableReply.parseTerminatedPrefix(parsingBuffer->c_str(), parsingBuffer->contentSize()))
        return true;
 
    // TODO: Optimize by checking memory as well. For simplicity sake, we
    // continue on the disk-reading path, but readFromMemory() can give us the
    // missing header bytes immediately if a concurrent request put those bytes
    // into memory while we were waiting for our disk response.
    scheduleDiskRead();
    return false;
}
 
void
store_client::skipHttpHeadersFromDisk()
{
    const auto hdr_sz = entry->mem_obj->baseReply().hdr_sz;
    Assure(hdr_sz > 0); // all on-disk responses have HTTP headers
    if (Less(parsingBuffer->contentSize(), hdr_sz)) {
        debugs(90, 5, "discovered " << hdr_sz << "-byte HTTP headers in memory after reading some of them from disk: " << *parsingBuffer);
        parsingBuffer->consume(parsingBuffer->contentSize()); // skip loaded HTTP header prefix
    } else {
        parsingBuffer->consume(hdr_sz); // skip loaded HTTP headers
        const auto httpBodyBytesAfterHeader = parsingBuffer->contentSize(); // may be zero
        Assure(httpBodyBytesAfterHeader <= copyInto.length);
        debugs(90, 5, "read HTTP body prefix: " << httpBodyBytesAfterHeader);
    }
}
 
void
store_client::dumpStats(MemBuf * output, int clientNumber) const
{
    if (_callback.pending())
        return;
 
    output->appendf("\tClient #%d, %p\n", clientNumber, this);
    output->appendf("\t\tcopy_offset: %" PRId64 "\n", copyInto.offset);
    output->appendf("\t\tcopy_size: %zu\n", copyInto.length);
    output->append("\t\tflags:", 8);
 
    if (flags.disk_io_pending)
        output->append(" disk_io_pending", 16);
 
    if (flags.store_copying)
        output->append(" store_copying", 14);
 
    if (_callback.notifier)
        output->append(" notifying", 10);
 
    output->append("\n",1);
}
 
bool
store_client::Callback::pending() const
{
    return callback_handler && !notifier;
}
 
store_client::Callback::Callback(STCB *function, void *data):
    callback_handler(function),
    cbData(data),
    codeContext(CodeContext::Current())
{
}
 
#if USE_DELAY_POOLS
int
store_client::bytesWanted() const
{
    // TODO: To avoid using stale copyInto, return zero if !_callback.pending()?
    return delayId.bytesWanted(0, copyInto.length);
}
 
void
store_client::setDelayId(DelayId delay_id)
{
    delayId = delay_id;
}
#endif