/*
- * DEBUG: section 47 Store Directory Routines
+ * Copyright (C) 1996-2015 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 47 Store Directory Routines */
+
#include "squid.h"
#include "cache_cf.h"
+#include "CollapsedForwarding.h"
#include "ConfigOption.h"
#include "DiskIO/DiskIOModule.h"
#include "DiskIO/DiskIOStrategy.h"
#include "DiskIO/ReadRequest.h"
#include "DiskIO/WriteRequest.h"
-#include "fs/rock/RockSwapDir.h"
-#include "fs/rock/RockIoState.h"
#include "fs/rock/RockIoRequests.h"
+#include "fs/rock/RockIoState.h"
#include "fs/rock/RockRebuild.h"
+#include "fs/rock/RockSwapDir.h"
#include "globals.h"
#include "ipc/mem/Pages.h"
#include "MemObject.h"
#include <cstdlib>
#include <iomanip>
+#include <limits>
#if HAVE_SYS_STAT_H
#include <sys/stat.h>
const int64_t Rock::SwapDir::HeaderSize = 16*1024;
-Rock::SwapDir::SwapDir(): ::SwapDir("rock"),
+Rock::SwapDir::SwapDir(): ::SwapDir("rock"),
slotSize(HeaderSize), filePath(NULL), map(NULL), io(NULL),
waitingForPage(NULL)
{
}
StoreSearch *
-Rock::SwapDir::search(String const url, HttpRequest *)
+Rock::SwapDir::search(String const, HttpRequest *)
{
assert(false);
return NULL; // XXX: implement
if (!slot)
return NULL;
- const Ipc::StoreMapAnchor::Basics &basics = slot->basics;
-
// create a brand new store entry and initialize it with stored basics
StoreEntry *e = new StoreEntry();
- e->lock_count = 0;
- e->swap_dirn = index;
- e->swap_filen = filen;
- e->swap_file_sz = basics.swap_file_sz;
- e->lastref = basics.lastref;
- e->timestamp = basics.timestamp;
- e->expires = basics.expires;
- e->lastmod = basics.lastmod;
- e->refcount = basics.refcount;
- e->flags = basics.flags;
- e->store_status = STORE_OK;
- e->setMemStatus(NOT_IN_MEMORY);
- e->swap_status = SWAPOUT_DONE;
- e->ping_status = PING_NONE;
- EBIT_SET(e->flags, ENTRY_CACHABLE);
- EBIT_CLR(e->flags, RELEASE_REQUEST);
- EBIT_CLR(e->flags, KEY_PRIVATE);
- EBIT_SET(e->flags, ENTRY_VALIDATED);
+ anchorEntry(*e, filen, *slot);
+
e->hashInsert(key);
trackReferences(*e);
// the disk entry remains open for reading, protected from modifications
}
+bool
+Rock::SwapDir::anchorCollapsed(StoreEntry &collapsed, bool &inSync)
+{
+ if (!map || !theFile || !theFile->canRead())
+ return false;
+
+ sfileno filen;
+ const Ipc::StoreMapAnchor *const slot = map->openForReading(
+ reinterpret_cast<cache_key*>(collapsed.key), filen);
+ if (!slot)
+ return false;
+
+ anchorEntry(collapsed, filen, *slot);
+ inSync = updateCollapsedWith(collapsed, *slot);
+ return true; // even if inSync is false
+}
+
+bool
+Rock::SwapDir::updateCollapsed(StoreEntry &collapsed)
+{
+ if (!map || !theFile || !theFile->canRead())
+ return false;
+
+ if (collapsed.swap_filen < 0) // no longer using a disk cache
+ return true;
+ assert(collapsed.swap_dirn == index);
+
+ const Ipc::StoreMapAnchor &s = map->readableEntry(collapsed.swap_filen);
+ return updateCollapsedWith(collapsed, s);
+}
+
+bool
+Rock::SwapDir::updateCollapsedWith(StoreEntry &collapsed, const Ipc::StoreMapAnchor &anchor)
+{
+ collapsed.swap_file_sz = anchor.basics.swap_file_sz;
+ return true;
+}
+
+void
+Rock::SwapDir::anchorEntry(StoreEntry &e, const sfileno filen, const Ipc::StoreMapAnchor &anchor)
+{
+ const Ipc::StoreMapAnchor::Basics &basics = anchor.basics;
+
+ e.swap_file_sz = basics.swap_file_sz;
+ e.lastref = basics.lastref;
+ e.timestamp = basics.timestamp;
+ e.expires = basics.expires;
+ e.lastmod = basics.lastmod;
+ e.refcount = basics.refcount;
+ e.flags = basics.flags;
+
+ if (anchor.complete()) {
+ e.store_status = STORE_OK;
+ e.swap_status = SWAPOUT_DONE;
+ } else {
+ e.store_status = STORE_PENDING;
+ e.swap_status = SWAPOUT_WRITING; // even though another worker writes?
+ }
+
+ e.ping_status = PING_NONE;
+
+ EBIT_CLR(e.flags, RELEASE_REQUEST);
+ EBIT_CLR(e.flags, KEY_PRIVATE);
+ EBIT_SET(e.flags, ENTRY_VALIDATED);
+
+ e.swap_dirn = index;
+ e.swap_filen = filen;
+}
+
void Rock::SwapDir::disconnect(StoreEntry &e)
{
assert(e.swap_dirn == index);
// do not rely on e.swap_status here because there is an async delay
// before it switches from SWAPOUT_WRITING to SWAPOUT_DONE.
- // since e has swap_filen, its slot is locked for either reading or writing
- map->abortIo(e.swap_filen);
- e.swap_dirn = -1;
- e.swap_filen = -1;
- e.swap_status = SWAPOUT_NONE;
+ // since e has swap_filen, its slot is locked for reading and/or writing
+ // but it is difficult to know whether THIS worker is reading or writing e,
+ // especially since we may switch from writing to reading. This code relies
+ // on Rock::IoState::writeableAnchor_ being set when we locked for writing.
+ if (e.mem_obj && e.mem_obj->swapout.sio != NULL &&
+ dynamic_cast<IoState&>(*e.mem_obj->swapout.sio).writeableAnchor_) {
+ map->abortWriting(e.swap_filen);
+ e.swap_dirn = -1;
+ e.swap_filen = -1;
+ e.swap_status = SWAPOUT_NONE;
+ dynamic_cast<IoState&>(*e.mem_obj->swapout.sio).writeableAnchor_ = NULL;
+ Store::Root().transientsAbandon(e); // broadcasts after the change
+ } else {
+ map->closeForReading(e.swap_filen);
+ e.swap_dirn = -1;
+ e.swap_filen = -1;
+ e.swap_status = SWAPOUT_NONE;
+ }
}
uint64_t
Rock::SwapDir::currentSize() const
{
const uint64_t spaceSize = !freeSlots ?
- maxSize() : (slotSize * freeSlots->size());
+ maxSize() : (slotSize * freeSlots->size());
// everything that is not free is in use
return maxSize() - spaceSize;
}
}
int64_t
-Rock::SwapDir::entryLimitAllowed() const
+Rock::SwapDir::slotLimitAbsolute() const
+{
+ // the max value is an invalid one; all values must be below the limit
+ assert(std::numeric_limits<Ipc::StoreMapSliceId>::max() ==
+ std::numeric_limits<SlotId>::max());
+ return std::numeric_limits<SlotId>::max();
+}
+
+int64_t
+Rock::SwapDir::slotLimitActual() const
{
- const int64_t eLimitLo = map ? map->entryLimit() : 0; // dynamic shrinking unsupported
- const int64_t eWanted = (maxSize() - HeaderSize)/slotSize;
- return min(max(eLimitLo, eWanted), entryLimitHigh());
+ const int64_t sWanted = (maxSize() - HeaderSize)/slotSize;
+ const int64_t sLimitLo = map ? map->sliceLimit() : 0; // dynamic shrinking unsupported
+ const int64_t sLimitHi = slotLimitAbsolute();
+ return min(max(sLimitLo, sWanted), sLimitHi);
}
-// TODO: encapsulate as a tool; identical to CossSwapDir::create()
+int64_t
+Rock::SwapDir::entryLimitActual() const
+{
+ return min(slotLimitActual(), entryLimitAbsolute());
+}
+
+// TODO: encapsulate as a tool
void
Rock::SwapDir::create()
{
// report Rock DB creation error and exit
void
-Rock::SwapDir::createError(const char *const msg) {
+Rock::SwapDir::createError(const char *const msg)
+{
debugs(47, DBG_CRITICAL, "ERROR: Failed to initialize Rock Store db in " <<
filePath << "; " << msg << " error: " << xstrerror());
fatal("Rock Store db creation error");
const int64_t slotSizeRoundingWaste = slotSize;
const int64_t maxRoundingWaste =
max(maxSizeRoundingWaste, slotSizeRoundingWaste);
- const int64_t usableDiskSize = diskOffset(entryLimitAllowed());
- const int64_t diskWasteSize = maxSize() - usableDiskSize;
- Must(diskWasteSize >= 0);
- // warn if maximum db size is not reachable due to sfileno limit
- if (entryLimitAllowed() == entryLimitHigh() &&
- diskWasteSize >= maxRoundingWaste) {
- debugs(47, DBG_CRITICAL, "Rock store cache_dir[" << index << "] '" << path << "':");
- debugs(47, DBG_CRITICAL, "\tmaximum number of entries: " << entryLimitAllowed());
- debugs(47, DBG_CRITICAL, "\tdb slot size: " << slotSize << " Bytes");
- debugs(47, DBG_CRITICAL, "\tmaximum db size: " << maxSize() << " Bytes");
- debugs(47, DBG_CRITICAL, "\tusable db size: " << usableDiskSize << " Bytes");
- debugs(47, DBG_CRITICAL, "\tdisk space waste: " << diskWasteSize << " Bytes");
- debugs(47, DBG_CRITICAL, "WARNING: Rock store config wastes space.");
+ // an entry consumes at least one slot; round up to reduce false warnings
+ const int64_t blockSize = static_cast<int64_t>(slotSize);
+ const int64_t maxObjSize = max(blockSize,
+ ((maxObjectSize()+blockSize-1)/blockSize)*blockSize);
+
+ // Does the "sfileno*max-size" limit match configured db capacity?
+ const double entriesMayOccupy = entryLimitAbsolute()*static_cast<double>(maxObjSize);
+ if (entriesMayOccupy + maxRoundingWaste < maxSize()) {
+ const int64_t diskWasteSize = maxSize() - static_cast<int64_t>(entriesMayOccupy);
+ debugs(47, DBG_CRITICAL, "WARNING: Rock cache_dir " << path << " wastes disk space due to entry limits:" <<
+ "\n\tconfigured db capacity: " << maxSize() << " bytes" <<
+ "\n\tconfigured db slot size: " << slotSize << " bytes" <<
+ "\n\tconfigured maximum entry size: " << maxObjectSize() << " bytes" <<
+ "\n\tmaximum number of cache_dir entries supported by Squid: " << entryLimitAbsolute() <<
+ "\n\tdisk space all entries may use: " << entriesMayOccupy << " bytes" <<
+ "\n\tdisk space wasted: " << diskWasteSize << " bytes");
+ }
+
+ // Does the "absolute slot count" limit match configured db capacity?
+ const double slotsMayOccupy = slotLimitAbsolute()*static_cast<double>(slotSize);
+ if (slotsMayOccupy + maxRoundingWaste < maxSize()) {
+ const int64_t diskWasteSize = maxSize() - static_cast<int64_t>(entriesMayOccupy);
+ debugs(47, DBG_CRITICAL, "WARNING: Rock cache_dir " << path << " wastes disk space due to slot limits:" <<
+ "\n\tconfigured db capacity: " << maxSize() << " bytes" <<
+ "\n\tconfigured db slot size: " << slotSize << " bytes" <<
+ "\n\tmaximum number of rock cache_dir slots supported by Squid: " << slotLimitAbsolute() <<
+ "\n\tdisk space all slots may use: " << slotsMayOccupy << " bytes" <<
+ "\n\tdisk space wasted: " << diskWasteSize << " bytes");
}
}
}
int64_t
-Rock::SwapDir::diskOffset(int filen) const
+Rock::SwapDir::diskOffset(const SlotId sid) const
{
- assert(filen >= 0);
- return HeaderSize + slotSize*filen;
+ assert(sid >= 0);
+ return HeaderSize + slotSize*sid;
}
int64_t
Rock::SwapDir::diskOffsetLimit() const
{
assert(map);
- return diskOffset(map->entryLimit());
-}
-
-int
-Rock::SwapDir::entryMaxPayloadSize() const
-{
- return slotSize - sizeof(DbCellHeader);
-}
-
-int
-Rock::SwapDir::entriesNeeded(const int64_t objSize) const
-{
- return (objSize + entryMaxPayloadSize() - 1) / entryMaxPayloadSize();
+ return diskOffset(map->sliceLimit());
}
bool
bool
Rock::SwapDir::validSlotId(const SlotId slotId) const
{
- return 0 <= slotId && slotId < entryLimitAllowed();
+ return 0 <= slotId && slotId < slotLimitActual();
}
void
-Rock::SwapDir::noteFreeMapSlice(const sfileno sliceId)
+Rock::SwapDir::noteFreeMapSlice(const Ipc::StoreMapSliceId sliceId)
{
Ipc::Mem::PageId pageId;
pageId.pool = index+1;
// The are two ways an entry can get swap_filen: our get() locked it for
// reading or our storeSwapOutStart() locked it for writing. Peeking at our
- // locked entry is safe, but no support for reading a filling entry.
+ // locked entry is safe, but no support for reading the entry we swap out.
const Ipc::StoreMapAnchor *slot = map->peekAtReader(e.swap_filen);
if (!slot)
return NULL; // we were writing afterall
sio->swap_filen);
assert(slot->sameKey(static_cast<const cache_key*>(e.key)));
- assert(slot->basics.swap_file_sz > 0);
- assert(slot->basics.swap_file_sz == e.swap_file_sz);
+ // For collapsed disk hits: e.swap_file_sz and slot->basics.swap_file_sz
+ // may still be zero and basics.swap_file_sz may grow.
+ assert(slot->basics.swap_file_sz >= e.swap_file_sz);
return sio;
}
xstrerror());
debugs(47, 2, "Rock cache_dir[" << index << "] limits: " <<
- std::setw(12) << maxSize() << " disk bytes and " <<
- std::setw(7) << map->entryLimit() << " entries");
+ std::setw(12) << maxSize() << " disk bytes, " <<
+ std::setw(7) << map->entryLimit() << " entries, and " <<
+ std::setw(7) << map->sliceLimit() << " slots");
rebuild();
}
}
void
-Rock::SwapDir::readCompleted(const char *buf, int rlen, int errflag, RefCount< ::ReadRequest> r)
+Rock::SwapDir::readCompleted(const char *, int rlen, int errflag, RefCount< ::ReadRequest> r)
{
ReadRequest *request = dynamic_cast<Rock::ReadRequest*>(r.getRaw());
assert(request);
if (errflag == DISK_OK && rlen > 0)
sio->offset_ += rlen;
- StoreIOState::STRCB *callb = sio->read.callback;
- assert(callb);
- sio->read.callback = NULL;
- void *cbdata;
- if (cbdataReferenceValidDone(sio->read.callback_data, &cbdata))
- callb(cbdata, r->buf, rlen, sio.getRaw());
+ sio->callReaderBack(r->buf, rlen);
}
void
-Rock::SwapDir::writeCompleted(int errflag, size_t rlen, RefCount< ::WriteRequest> r)
+Rock::SwapDir::writeCompleted(int errflag, size_t, RefCount< ::WriteRequest> r)
{
Rock::WriteRequest *request = dynamic_cast<Rock::WriteRequest*>(r.getRaw());
assert(request);
// quit if somebody called IoState::close() while we were waiting
if (!sio.stillWaiting()) {
debugs(79, 3, "ignoring closed entry " << sio.swap_filen);
+ noteFreeMapSlice(request->sidNext);
return;
}
+ // TODO: Fail if disk dropped one of the previous write requests.
+
if (errflag == DISK_OK) {
// do not increment sio.offset_ because we do it in sio->write()
- if (request->isLast) {
+
+ // finalize the shared slice info after writing slice contents to disk
+ Ipc::StoreMap::Slice &slice =
+ map->writeableSlice(sio.swap_filen, request->sidCurrent);
+ slice.size = request->len - sizeof(DbCellHeader);
+ slice.next = request->sidNext;
+
+ if (request->eof) {
+ assert(sio.e);
+ assert(sio.writeableAnchor_);
+ sio.e->swap_file_sz = sio.writeableAnchor_->basics.swap_file_sz =
+ sio.offset_;
+
// close, the entry gets the read lock
map->closeForWriting(sio.swap_filen, true);
+ sio.writeableAnchor_ = NULL;
sio.finishedWriting(errflag);
}
} else {
- writeError(sio.swap_filen);
+ noteFreeMapSlice(request->sidNext);
+
+ writeError(*sio.e);
sio.finishedWriting(errflag);
// and hope that Core will call disconnect() to close the map entry
}
+
+ CollapsedForwarding::Broadcast(*sio.e);
}
void
-Rock::SwapDir::writeError(const sfileno fileno)
+Rock::SwapDir::writeError(StoreEntry &e)
{
// Do not abortWriting here. The entry should keep the write lock
// instead of losing association with the store and confusing core.
- map->freeEntry(fileno); // will mark as unusable, just in case
+ map->freeEntry(e.swap_filen); // will mark as unusable, just in case
+
+ Store::Root().transientsAbandon(e);
+
// All callers must also call IoState callback, to propagate the error.
}
disconnect(e);
}
+void
+Rock::SwapDir::markForUnlink(StoreEntry &e)
+{
+ debugs(47, 5, e);
+ map->freeEntry(e.swap_filen);
+}
+
void
Rock::SwapDir::trackReferences(StoreEntry &e)
{
currentSize() / 1024.0,
Math::doublePercent(currentSize(), maxSize()));
- if (map) {
- const int limit = map->entryLimit();
- storeAppendPrintf(&e, "Maximum entries: %9d\n", limit);
- if (limit > 0) {
- const int entryCount = map->entryCount();
- storeAppendPrintf(&e, "Current entries: %9d %.2f%%\n",
- entryCount, (100.0 * entryCount / limit));
-
- const unsigned int slotsFree = !freeSlots ? 0 : freeSlots->size();
- if (slotsFree <= static_cast<const unsigned int>(limit)) {
- const int usedSlots = limit - static_cast<const int>(slotsFree);
- storeAppendPrintf(&e, "Used slots: %9d %.2f%%\n",
- usedSlots, (100.0 * usedSlots / limit));
- }
- if (limit < 100) { // XXX: otherwise too expensive to count
- Ipc::ReadWriteLockStats stats;
- map->updateStats(stats);
- stats.dump(e);
- }
+ const int entryLimit = entryLimitActual();
+ const int slotLimit = slotLimitActual();
+ storeAppendPrintf(&e, "Maximum entries: %9d\n", entryLimit);
+ if (map && entryLimit > 0) {
+ const int entryCount = map->entryCount();
+ storeAppendPrintf(&e, "Current entries: %9d %.2f%%\n",
+ entryCount, (100.0 * entryCount / entryLimit));
+ }
+
+ storeAppendPrintf(&e, "Maximum slots: %9d\n", slotLimit);
+ if (map && slotLimit > 0) {
+ const unsigned int slotsFree = !freeSlots ? 0 : freeSlots->size();
+ if (slotsFree <= static_cast<const unsigned int>(slotLimit)) {
+ const int usedSlots = slotLimit - static_cast<const int>(slotsFree);
+ storeAppendPrintf(&e, "Used slots: %9d %.2f%%\n",
+ usedSlots, (100.0 * usedSlots / slotLimit));
+ }
+ if (slotLimit < 100) { // XXX: otherwise too expensive to count
+ Ipc::ReadWriteLockStats stats;
+ map->updateStats(stats);
+ stats.dump(e);
}
}
}
-const char *
-Rock::SwapDir::inodeMapPath() const {
- static String inodesPath;
- inodesPath = path;
- inodesPath.append("_inodes");
- return inodesPath.termedBuf();
+SBuf
+Rock::SwapDir::inodeMapPath() const
+{
+ return Ipc::Mem::Segment::Name(SBuf(path), "map");
}
const char *
-Rock::SwapDir::freeSlotsPath() const {
+Rock::SwapDir::freeSlotsPath() const
+{
static String spacesPath;
spacesPath = path;
spacesPath.append("_spaces");
namespace Rock
{
-RunnerRegistrationEntry(rrAfterConfig, SwapDirRr);
+RunnerRegistrationEntry(SwapDirRr);
}
-void Rock::SwapDirRr::create(const RunnerRegistry &)
+void Rock::SwapDirRr::create()
{
Must(mapOwners.empty() && freeSlotsOwners.empty());
for (int i = 0; i < Config.cacheSwap.n_configured; ++i) {
if (const Rock::SwapDir *const sd = dynamic_cast<Rock::SwapDir *>(INDEXSD(i))) {
- const int64_t capacity = sd->entryLimitAllowed();
+ const int64_t capacity = sd->slotLimitActual();
SwapDir::DirMap::Owner *const mapOwner =
SwapDir::DirMap::Init(sd->inodeMapPath(), capacity);
mapOwners.push_back(mapOwner);
- // XXX: remove pool id and counters from PageStack
+ // TODO: somehow remove pool id and counters from PageStack?
Ipc::Mem::Owner<Ipc::Mem::PageStack> *const freeSlotsOwner =
shm_new(Ipc::Mem::PageStack)(sd->freeSlotsPath(),
- i+1, capacity,
- sizeof(DbCellHeader));
+ i+1, capacity, 0);
freeSlotsOwners.push_back(freeSlotsOwner);
- // XXX: add method to initialize PageStack with no free pages
+ // TODO: add method to initialize PageStack with no free pages
while (true) {
Ipc::Mem::PageId pageId;
if (!freeSlotsOwner->object()->pop(pageId))
delete freeSlotsOwners[i];
}
}
+
projects / thirdparty / squid.git / blobdiff