src/fs/rock/RockSwapDir.cc

   1 /*
   2  * $Id$
   3  *
   4  * DEBUG: section 47    Store Directory Routines
   5  */
   6
   7 #include "squid.h"
   8 #include "cache_cf.h"
   9 #include "ConfigOption.h"
  10 #include "DiskIO/DiskIOModule.h"
  11 #include "DiskIO/DiskIOStrategy.h"
  12 #include "DiskIO/ReadRequest.h"
  13 #include "DiskIO/WriteRequest.h"
  14 #include "fs/rock/RockSwapDir.h"
  15 #include "fs/rock/RockIoState.h"
  16 #include "fs/rock/RockIoRequests.h"
  17 #include "fs/rock/RockRebuild.h"
  18 #include "globals.h"
  19 #include "ipc/mem/Pages.h"
  20 #include "MemObject.h"
  21 #include "Parsing.h"
  22 #include "protos.h"
  23 #include "SquidMath.h"
  24 #include "tools.h"
  25
  26 #include <cstdlib>
  27 #include <iomanip>
  28
  29 #if HAVE_SYS_STAT_H
  30 #include <sys/stat.h>
  31 #endif
  32
  33 const int64_t Rock::SwapDir::HeaderSize = 16*1024;
  34
  35 Rock::SwapDir::SwapDir(): ::SwapDir("rock"), filePath(NULL), io(NULL), map(NULL)
  36 {
  37 }
  38
  39 Rock::SwapDir::~SwapDir()
  40 {
  41     delete io;
  42     delete map;
  43     safe_free(filePath);
  44 }
  45
  46 StoreSearch *
  47 Rock::SwapDir::search(String const url, HttpRequest *)
  48 {
  49     assert(false);
  50     return NULL; // XXX: implement
  51 }
  52
  53 void
  54 Rock::SwapDir::get(String const key, STOREGETCLIENT cb, void *data)
  55 {
  56     ::SwapDir::get(key, cb, data);
  57 }
  58
  59 // called when Squid core needs a StoreEntry with a given key
  60 StoreEntry *
  61 Rock::SwapDir::get(const cache_key *key)
  62 {
  63     if (!map || !theFile || !theFile->canRead())
  64         return NULL;
  65
  66     sfileno filen;
  67     const Ipc::StoreMapSlot *const slot = map->openForReading(key, filen);
  68     if (!slot)
  69         return NULL;
  70
  71     const Ipc::StoreMapSlot::Basics &basics = slot->basics;
  72
  73     // create a brand new store entry and initialize it with stored basics
  74     StoreEntry *e = new StoreEntry();
  75     e->lock_count = 0;
  76     e->swap_dirn = index;
  77     e->swap_filen = filen;
  78     e->swap_file_sz = basics.swap_file_sz;
  79     e->lastref = basics.lastref;
  80     e->timestamp = basics.timestamp;
  81     e->expires = basics.expires;
  82     e->lastmod = basics.lastmod;
  83     e->refcount = basics.refcount;
  84     e->flags = basics.flags;
  85     e->store_status = STORE_OK;
  86     e->setMemStatus(NOT_IN_MEMORY);
  87     e->swap_status = SWAPOUT_DONE;
  88     e->ping_status = PING_NONE;
  89     EBIT_SET(e->flags, ENTRY_CACHABLE);
  90     EBIT_CLR(e->flags, RELEASE_REQUEST);
  91     EBIT_CLR(e->flags, KEY_PRIVATE);
  92     EBIT_SET(e->flags, ENTRY_VALIDATED);
  93     e->hashInsert(key);
  94     trackReferences(*e);
  95
  96     return e;
  97     // the disk entry remains open for reading, protected from modifications
  98 }
  99
 100 void Rock::SwapDir::disconnect(StoreEntry &e)
 101 {
 102     assert(e.swap_dirn == index);
 103     assert(e.swap_filen >= 0);
 104     // cannot have SWAPOUT_NONE entry with swap_filen >= 0
 105     assert(e.swap_status != SWAPOUT_NONE);
 106
 107     // do not rely on e.swap_status here because there is an async delay
 108     // before it switches from SWAPOUT_WRITING to SWAPOUT_DONE.
 109
 110     // since e has swap_filen, its slot is locked for either reading or writing
 111     map->abortIo(e.swap_filen);
 112     e.swap_dirn = -1;
 113     e.swap_filen = -1;
 114     e.swap_status = SWAPOUT_NONE;
 115 }
 116
 117 uint64_t
 118 Rock::SwapDir::currentSize() const
 119 {
 120     return HeaderSize + max_objsize * currentCount();
 121 }
 122
 123 uint64_t
 124 Rock::SwapDir::currentCount() const
 125 {
 126     return map ? map->entryCount() : 0;
 127 }
 128
 129 /// In SMP mode only the disker process reports stats to avoid
 130 /// counting the same stats by multiple processes.
 131 bool
 132 Rock::SwapDir::doReportStat() const
 133 {
 134     return ::SwapDir::doReportStat() && (!UsingSmp() || IamDiskProcess());
 135 }
 136
 137 void
 138 Rock::SwapDir::swappedOut(const StoreEntry &)
 139 {
 140     // stats are not stored but computed when needed
 141 }
 142
 143 int64_t
 144 Rock::SwapDir::entryLimitAllowed() const
 145 {
 146     const int64_t eLimitLo = map ? map->entryLimit() : 0; // dynamic shrinking unsupported
 147     const int64_t eWanted = (maxSize() - HeaderSize)/maxObjectSize();
 148     return min(max(eLimitLo, eWanted), entryLimitHigh());
 149 }
 150
 151 // TODO: encapsulate as a tool; identical to CossSwapDir::create()
 152 void
 153 Rock::SwapDir::create()
 154 {
 155     assert(path);
 156     assert(filePath);
 157
 158     if (UsingSmp() && !IamDiskProcess()) {
 159         debugs (47,3, HERE << "disker will create in " << path);
 160         return;
 161     }
 162
 163     debugs (47,3, HERE << "creating in " << path);
 164
 165     struct stat swap_sb;
 166     if (::stat(path, &swap_sb) < 0) {
 167         debugs (47, DBG_IMPORTANT, "Creating Rock db directory: " << path);
 168         const int res = mkdir(path, 0700);
 169         if (res != 0) {
 170             debugs(47, DBG_CRITICAL, "Failed to create Rock db dir " << path <<
 171                    ": " << xstrerror());
 172             fatal("Rock Store db creation error");
 173         }
 174     }
 175
 176 #if SLOWLY_FILL_WITH_ZEROS
 177     char block[1024];
 178     Must(maxSize() % sizeof(block) == 0);
 179     memset(block, '\0', sizeof(block));
 180
 181     const int swap = open(filePath, O_WRONLY|O_CREAT|O_TRUNC|O_BINARY, 0600);
 182     for (off_t offset = 0; offset < maxSize(); offset += sizeof(block)) {
 183         if (write(swap, block, sizeof(block)) != sizeof(block)) {
 184             debugs(47, DBG_CRITICAL, "ERROR: Failed to create Rock Store db in " << filePath <<
 185                    ": " << xstrerror());
 186             fatal("Rock Store db creation error");
 187         }
 188     }
 189     close(swap);
 190 #else
 191     const int swap = open(filePath, O_WRONLY|O_CREAT|O_TRUNC|O_BINARY, 0600);
 192     if (swap < 0) {
 193         debugs(47, DBG_CRITICAL, "ERROR: Failed to initialize Rock Store db in " << filePath <<
 194                "; create error: " << xstrerror());
 195         fatal("Rock Store db creation error");
 196     }
 197
 198     if (ftruncate(swap, maxSize()) != 0) {
 199         debugs(47, DBG_CRITICAL, "ERROR: Failed to initialize Rock Store db in " << filePath <<
 200                "; truncate error: " << xstrerror());
 201         fatal("Rock Store db creation error");
 202     }
 203
 204     char header[HeaderSize];
 205     memset(header, '\0', sizeof(header));
 206     if (write(swap, header, sizeof(header)) != sizeof(header)) {
 207         debugs(47, DBG_CRITICAL, "ERROR: Failed to initialize Rock Store db in " << filePath <<
 208                "; write error: " << xstrerror());
 209         fatal("Rock Store db initialization error");
 210     }
 211     close(swap);
 212 #endif
 213 }
 214
 215 void
 216 Rock::SwapDir::init()
 217 {
 218     debugs(47,2, HERE);
 219
 220     // XXX: SwapDirs aren't refcounted. We make IORequestor calls, which
 221     // are refcounted. We up our count once to avoid implicit delete's.
 222     RefCountReference();
 223
 224     Must(!map);
 225     map = new DirMap(path);
 226
 227     const char *ioModule = needsDiskStrand() ? "IpcIo" : "Blocking";
 228     if (DiskIOModule *m = DiskIOModule::Find(ioModule)) {
 229         debugs(47,2, HERE << "Using DiskIO module: " << ioModule);
 230         io = m->createStrategy();
 231         io->init();
 232     } else {
 233         debugs(47, DBG_CRITICAL, "FATAL: Rock store is missing DiskIO module: " <<
 234                ioModule);
 235         fatal("Rock Store missing a required DiskIO module");
 236     }
 237
 238     theFile = io->newFile(filePath);
 239     theFile->configure(fileConfig);
 240     theFile->open(O_RDWR, 0644, this);
 241
 242     // Increment early. Otherwise, if one SwapDir finishes rebuild before
 243     // others start, storeRebuildComplete() will think the rebuild is over!
 244     // TODO: move store_dirs_rebuilding hack to store modules that need it.
 245     ++StoreController::store_dirs_rebuilding;
 246 }
 247
 248 bool
 249 Rock::SwapDir::needsDiskStrand() const
 250 {
 251     const bool wontEvenWorkWithoutDisker = Config.workers > 1;
 252     const bool wouldWorkBetterWithDisker = DiskIOModule::Find("IpcIo");
 253     return InDaemonMode() && (wontEvenWorkWithoutDisker ||
 254                               wouldWorkBetterWithDisker);
 255 }
 256
 257 void
 258 Rock::SwapDir::parse(int anIndex, char *aPath)
 259 {
 260     index = anIndex;
 261
 262     path = xstrdup(aPath);
 263
 264     // cache store is located at path/db
 265     String fname(path);
 266     fname.append("/rock");
 267     filePath = xstrdup(fname.termedBuf());
 268
 269     parseSize(false);
 270     parseOptions(0);
 271
 272     // Current openForWriting() code overwrites the old slot if needed
 273     // and possible, so proactively removing old slots is probably useless.
 274     assert(!repl); // repl = createRemovalPolicy(Config.replPolicy);
 275
 276     validateOptions();
 277 }
 278
 279 void
 280 Rock::SwapDir::reconfigure()
 281 {
 282     parseSize(true);
 283     parseOptions(1);
 284     // TODO: can we reconfigure the replacement policy (repl)?
 285     validateOptions();
 286 }
 287
 288 /// parse maximum db disk size
 289 void
 290 Rock::SwapDir::parseSize(const bool reconfiguring)
 291 {
 292     const int i = GetInteger();
 293     if (i < 0)
 294         fatal("negative Rock cache_dir size value");
 295     const uint64_t new_max_size =
 296         static_cast<uint64_t>(i) << 20; // MBytes to Bytes
 297     if (!reconfiguring)
 298         max_size = new_max_size;
 299     else if (new_max_size != max_size) {
 300         debugs(3, DBG_IMPORTANT, "WARNING: cache_dir '" << path << "' size "
 301                "cannot be changed dynamically, value left unchanged (" <<
 302                (max_size >> 20) << " MB)");
 303     }
 304 }
 305
 306 ConfigOption *
 307 Rock::SwapDir::getOptionTree() const
 308 {
 309     ConfigOptionVector *vector = dynamic_cast<ConfigOptionVector*>(::SwapDir::getOptionTree());
 310     assert(vector);
 311     vector->options.push_back(new ConfigOptionAdapter<SwapDir>(*const_cast<SwapDir *>(this), &SwapDir::parseTimeOption, &SwapDir::dumpTimeOption));
 312     vector->options.push_back(new ConfigOptionAdapter<SwapDir>(*const_cast<SwapDir *>(this), &SwapDir::parseRateOption, &SwapDir::dumpRateOption));
 313     return vector;
 314 }
 315
 316 bool
 317 Rock::SwapDir::allowOptionReconfigure(const char *const option) const
 318 {
 319     return strcmp(option, "max-size") != 0 &&
 320            ::SwapDir::allowOptionReconfigure(option);
 321 }
 322
 323 /// parses time-specific options; mimics ::SwapDir::optionObjectSizeParse()
 324 bool
 325 Rock::SwapDir::parseTimeOption(char const *option, const char *value, int reconfiguring)
 326 {
 327     // TODO: ::SwapDir or, better, Config should provide time-parsing routines,
 328     // including time unit handling. Same for size.
 329
 330     time_msec_t *storedTime;
 331     if (strcmp(option, "swap-timeout") == 0)
 332         storedTime = &fileConfig.ioTimeout;
 333     else
 334         return false;
 335
 336     if (!value)
 337         self_destruct();
 338
 339     // TODO: handle time units and detect parsing errors better
 340     const int64_t parsedValue = strtoll(value, NULL, 10);
 341     if (parsedValue < 0) {
 342         debugs(3, DBG_CRITICAL, "FATAL: cache_dir " << path << ' ' << option << " must not be negative but is: " << parsedValue);
 343         self_destruct();
 344     }
 345
 346     const time_msec_t newTime = static_cast<time_msec_t>(parsedValue);
 347
 348     if (!reconfiguring)
 349         *storedTime = newTime;
 350     else if (*storedTime != newTime) {
 351         debugs(3, DBG_IMPORTANT, "WARNING: cache_dir " << path << ' ' << option
 352                << " cannot be changed dynamically, value left unchanged: " <<
 353                *storedTime);
 354     }
 355
 356     return true;
 357 }
 358
 359 /// reports time-specific options; mimics ::SwapDir::optionObjectSizeDump()
 360 void
 361 Rock::SwapDir::dumpTimeOption(StoreEntry * e) const
 362 {
 363     if (fileConfig.ioTimeout)
 364         storeAppendPrintf(e, " swap-timeout=%" PRId64,
 365                           static_cast<int64_t>(fileConfig.ioTimeout));
 366 }
 367
 368 /// parses rate-specific options; mimics ::SwapDir::optionObjectSizeParse()
 369 bool
 370 Rock::SwapDir::parseRateOption(char const *option, const char *value, int isaReconfig)
 371 {
 372     int *storedRate;
 373     if (strcmp(option, "max-swap-rate") == 0)
 374         storedRate = &fileConfig.ioRate;
 375     else
 376         return false;
 377
 378     if (!value)
 379         self_destruct();
 380
 381     // TODO: handle time units and detect parsing errors better
 382     const int64_t parsedValue = strtoll(value, NULL, 10);
 383     if (parsedValue < 0) {
 384         debugs(3, DBG_CRITICAL, "FATAL: cache_dir " << path << ' ' << option << " must not be negative but is: " << parsedValue);
 385         self_destruct();
 386     }
 387
 388     const int newRate = static_cast<int>(parsedValue);
 389
 390     if (newRate < 0) {
 391         debugs(3, DBG_CRITICAL, "FATAL: cache_dir " << path << ' ' << option << " must not be negative but is: " << newRate);
 392         self_destruct();
 393     }
 394
 395     if (!isaReconfig)
 396         *storedRate = newRate;
 397     else if (*storedRate != newRate) {
 398         debugs(3, DBG_IMPORTANT, "WARNING: cache_dir " << path << ' ' << option
 399                << " cannot be changed dynamically, value left unchanged: " <<
 400                *storedRate);
 401     }
 402
 403     return true;
 404 }
 405
 406 /// reports rate-specific options; mimics ::SwapDir::optionObjectSizeDump()
 407 void
 408 Rock::SwapDir::dumpRateOption(StoreEntry * e) const
 409 {
 410     if (fileConfig.ioRate >= 0)
 411         storeAppendPrintf(e, " max-swap-rate=%d", fileConfig.ioRate);
 412 }
 413
 414 /// check the results of the configuration; only level-0 debugging works here
 415 void
 416 Rock::SwapDir::validateOptions()
 417 {
 418     if (max_objsize <= 0)
 419         fatal("Rock store requires a positive max-size");
 420
 421     const int64_t maxSizeRoundingWaste = 1024 * 1024; // size is configured in MB
 422     const int64_t maxObjectSizeRoundingWaste = maxObjectSize();
 423     const int64_t maxRoundingWaste =
 424         max(maxSizeRoundingWaste, maxObjectSizeRoundingWaste);
 425     const int64_t usableDiskSize = diskOffset(entryLimitAllowed());
 426     const int64_t diskWasteSize = maxSize() - usableDiskSize;
 427     Must(diskWasteSize >= 0);
 428
 429     // warn if maximum db size is not reachable due to sfileno limit
 430     if (entryLimitAllowed() == entryLimitHigh() &&
 431             diskWasteSize >= maxRoundingWaste) {
 432         debugs(47, DBG_CRITICAL, "Rock store cache_dir[" << index << "] '" << path << "':");
 433         debugs(47, DBG_CRITICAL, "\tmaximum number of entries: " << entryLimitAllowed());
 434         debugs(47, DBG_CRITICAL, "\tmaximum object size: " << maxObjectSize() << " Bytes");
 435         debugs(47, DBG_CRITICAL, "\tmaximum db size: " << maxSize() << " Bytes");
 436         debugs(47, DBG_CRITICAL, "\tusable db size:  " << usableDiskSize << " Bytes");
 437         debugs(47, DBG_CRITICAL, "\tdisk space waste: " << diskWasteSize << " Bytes");
 438         debugs(47, DBG_CRITICAL, "WARNING: Rock store config wastes space.");
 439     }
 440 }
 441
 442 void
 443 Rock::SwapDir::rebuild()
 444 {
 445     //++StoreController::store_dirs_rebuilding; // see Rock::SwapDir::init()
 446     AsyncJob::Start(new Rebuild(this));
 447 }
 448
 449 /* Add a new object to the cache with empty memory copy and pointer to disk
 450  * use to rebuild store from disk. Based on UFSSwapDir::addDiskRestore */
 451 bool
 452 Rock::SwapDir::addEntry(const int filen, const DbCellHeader &header, const StoreEntry &from)
 453 {
 454     debugs(47, 8, HERE << &from << ' ' << from.getMD5Text() <<
 455            ", filen="<< std::setfill('0') << std::hex << std::uppercase <<
 456            std::setw(8) << filen);
 457
 458     sfileno newLocation = 0;
 459     if (Ipc::StoreMapSlot *slot = map->openForWriting(reinterpret_cast<const cache_key *>(from.key), newLocation)) {
 460         if (filen == newLocation) {
 461             slot->set(from);
 462             map->extras(filen) = header;
 463         } // else some other, newer entry got into our cell
 464         map->closeForWriting(newLocation, false);
 465         return filen == newLocation;
 466     }
 467
 468     return false;
 469 }
 470
 471 bool
 472 Rock::SwapDir::canStore(const StoreEntry &e, int64_t diskSpaceNeeded, int &load) const
 473 {
 474     if (!::SwapDir::canStore(e, sizeof(DbCellHeader)+diskSpaceNeeded, load))
 475         return false;
 476
 477     if (!theFile || !theFile->canWrite())
 478         return false;
 479
 480     if (!map)
 481         return false;
 482
 483     // Do not start I/O transaction if there are less than 10% free pages left.
 484     // TODO: reserve page instead
 485     if (needsDiskStrand() &&
 486             Ipc::Mem::PageLevel(Ipc::Mem::PageId::ioPage) >= 0.9 * Ipc::Mem::PageLimit(Ipc::Mem::PageId::ioPage)) {
 487         debugs(47, 5, HERE << "too few shared pages for IPC I/O left");
 488         return false;
 489     }
 490
 491     if (io->shedLoad())
 492         return false;
 493
 494     load = io->load();
 495     return true;
 496 }
 497
 498 StoreIOState::Pointer
 499 Rock::SwapDir::createStoreIO(StoreEntry &e, StoreIOState::STFNCB *cbFile, StoreIOState::STIOCB *cbIo, void *data)
 500 {
 501     if (!theFile || theFile->error()) {
 502         debugs(47,4, HERE << theFile);
 503         return NULL;
 504     }
 505
 506     // compute payload size for our cell header, using StoreEntry info
 507     // careful: e.objectLen() may still be negative here
 508     const int64_t expectedReplySize = e.mem_obj->expectedReplySize();
 509     assert(expectedReplySize >= 0); // must know to prevent cell overflows
 510     assert(e.mem_obj->swap_hdr_sz > 0);
 511     DbCellHeader header;
 512     header.payloadSize = e.mem_obj->swap_hdr_sz + expectedReplySize;
 513     const int64_t payloadEnd = sizeof(DbCellHeader) + header.payloadSize;
 514     assert(payloadEnd <= max_objsize);
 515
 516     sfileno filen;
 517     Ipc::StoreMapSlot *const slot =
 518         map->openForWriting(reinterpret_cast<const cache_key *>(e.key), filen);
 519     if (!slot) {
 520         debugs(47, 5, HERE << "map->add failed");
 521         return NULL;
 522     }
 523     e.swap_file_sz = header.payloadSize; // and will be copied to the map
 524     slot->set(e);
 525     map->extras(filen) = header;
 526
 527     // XXX: We rely on our caller, storeSwapOutStart(), to set e.fileno.
 528     // If that does not happen, the entry will not decrement the read level!
 529
 530     IoState *sio = new IoState(this, &e, cbFile, cbIo, data);
 531
 532     sio->swap_dirn = index;
 533     sio->swap_filen = filen;
 534     sio->payloadEnd = payloadEnd;
 535     sio->diskOffset = diskOffset(sio->swap_filen);
 536
 537     debugs(47,5, HERE << "dir " << index << " created new filen " <<
 538            std::setfill('0') << std::hex << std::uppercase << std::setw(8) <<
 539            sio->swap_filen << std::dec << " at " << sio->diskOffset);
 540
 541     assert(sio->diskOffset + payloadEnd <= diskOffsetLimit());
 542
 543     sio->file(theFile);
 544
 545     trackReferences(e);
 546     return sio;
 547 }
 548
 549 int64_t
 550 Rock::SwapDir::diskOffset(int filen) const
 551 {
 552     assert(filen >= 0);
 553     return HeaderSize + max_objsize*filen;
 554 }
 555
 556 int64_t
 557 Rock::SwapDir::diskOffsetLimit() const
 558 {
 559     assert(map);
 560     return diskOffset(map->entryLimit());
 561 }
 562
 563 // tries to open an old or being-written-to entry with swap_filen for reading
 564 StoreIOState::Pointer
 565 Rock::SwapDir::openStoreIO(StoreEntry &e, StoreIOState::STFNCB *cbFile, StoreIOState::STIOCB *cbIo, void *data)
 566 {
 567     if (!theFile || theFile->error()) {
 568         debugs(47,4, HERE << theFile);
 569         return NULL;
 570     }
 571
 572     if (e.swap_filen < 0) {
 573         debugs(47,4, HERE << e);
 574         return NULL;
 575     }
 576
 577     // Do not start I/O transaction if there are less than 10% free pages left.
 578     // TODO: reserve page instead
 579     if (needsDiskStrand() &&
 580             Ipc::Mem::PageLevel(Ipc::Mem::PageId::ioPage) >= 0.9 * Ipc::Mem::PageLimit(Ipc::Mem::PageId::ioPage)) {
 581         debugs(47, 5, HERE << "too few shared pages for IPC I/O left");
 582         return NULL;
 583     }
 584
 585     // The are two ways an entry can get swap_filen: our get() locked it for
 586     // reading or our storeSwapOutStart() locked it for writing. Peeking at our
 587     // locked entry is safe, but no support for reading a filling entry.
 588     const Ipc::StoreMapSlot *slot = map->peekAtReader(e.swap_filen);
 589     if (!slot)
 590         return NULL; // we were writing afterall
 591
 592     IoState *sio = new IoState(this, &e, cbFile, cbIo, data);
 593
 594     sio->swap_dirn = index;
 595     sio->swap_filen = e.swap_filen;
 596     sio->payloadEnd = sizeof(DbCellHeader) + map->extras(e.swap_filen).payloadSize;
 597     assert(sio->payloadEnd <= max_objsize); // the payload fits the slot
 598
 599     debugs(47,5, HERE << "dir " << index << " has old filen: " <<
 600            std::setfill('0') << std::hex << std::uppercase << std::setw(8) <<
 601            sio->swap_filen);
 602
 603     assert(slot->basics.swap_file_sz > 0);
 604     assert(slot->basics.swap_file_sz == e.swap_file_sz);
 605
 606     sio->diskOffset = diskOffset(sio->swap_filen);
 607     assert(sio->diskOffset + sio->payloadEnd <= diskOffsetLimit());
 608
 609     sio->file(theFile);
 610     return sio;
 611 }
 612
 613 void
 614 Rock::SwapDir::ioCompletedNotification()
 615 {
 616     if (!theFile)
 617         fatalf("Rock cache_dir failed to initialize db file: %s", filePath);
 618
 619     if (theFile->error())
 620         fatalf("Rock cache_dir at %s failed to open db file: %s", filePath,
 621                xstrerror());
 622
 623     debugs(47, 2, "Rock cache_dir[" << index << "] limits: " <<
 624            std::setw(12) << maxSize() << " disk bytes and " <<
 625            std::setw(7) << map->entryLimit() << " entries");
 626
 627     rebuild();
 628 }
 629
 630 void
 631 Rock::SwapDir::closeCompleted()
 632 {
 633     theFile = NULL;
 634 }
 635
 636 void
 637 Rock::SwapDir::readCompleted(const char *buf, int rlen, int errflag, RefCount< ::ReadRequest> r)
 638 {
 639     ReadRequest *request = dynamic_cast<Rock::ReadRequest*>(r.getRaw());
 640     assert(request);
 641     IoState::Pointer sio = request->sio;
 642
 643     if (errflag == DISK_OK && rlen > 0)
 644         sio->offset_ += rlen;
 645     assert(sio->diskOffset + sio->offset_ <= diskOffsetLimit()); // post-factum
 646
 647     StoreIOState::STRCB *callback = sio->read.callback;
 648     assert(callback);
 649     sio->read.callback = NULL;
 650     void *cbdata;
 651     if (cbdataReferenceValidDone(sio->read.callback_data, &cbdata))
 652         callback(cbdata, r->buf, rlen, sio.getRaw());
 653 }
 654
 655 void
 656 Rock::SwapDir::writeCompleted(int errflag, size_t rlen, RefCount< ::WriteRequest> r)
 657 {
 658     Rock::WriteRequest *request = dynamic_cast<Rock::WriteRequest*>(r.getRaw());
 659     assert(request);
 660     assert(request->sio !=  NULL);
 661     IoState &sio = *request->sio;
 662
 663     if (errflag == DISK_OK) {
 664         // close, assuming we only write once; the entry gets the read lock
 665         map->closeForWriting(sio.swap_filen, true);
 666         // do not increment sio.offset_ because we do it in sio->write()
 667     } else {
 668         // Do not abortWriting here. The entry should keep the write lock
 669         // instead of losing association with the store and confusing core.
 670         map->free(sio.swap_filen); // will mark as unusable, just in case
 671     }
 672
 673     assert(sio.diskOffset + sio.offset_ <= diskOffsetLimit()); // post-factum
 674
 675     sio.finishedWriting(errflag);
 676 }
 677
 678 bool
 679 Rock::SwapDir::full() const
 680 {
 681     return map && map->full();
 682 }
 683
 684 // storeSwapOutFileClosed calls this nethod on DISK_NO_SPACE_LEFT,
 685 // but it should not happen for us
 686 void
 687 Rock::SwapDir::diskFull()
 688 {
 689     debugs(20, DBG_IMPORTANT, "BUG: No space left with rock cache_dir: " <<
 690            filePath);
 691 }
 692
 693 /// purge while full(); it should be sufficient to purge just one
 694 void
 695 Rock::SwapDir::maintain()
 696 {
 697     debugs(47,3, HERE << "cache_dir[" << index << "] guards: " <<
 698            !repl << !map << !full() << StoreController::store_dirs_rebuilding);
 699
 700     if (!repl)
 701         return; // no means (cannot find a victim)
 702
 703     if (!map)
 704         return; // no victims (yet)
 705
 706     if (!full())
 707         return; // no need (to find a victim)
 708
 709     // XXX: UFSSwapDir::maintain says we must quit during rebuild
 710     if (StoreController::store_dirs_rebuilding)
 711         return;
 712
 713     debugs(47,3, HERE << "cache_dir[" << index << "] state: " << map->full() <<
 714            ' ' << currentSize() << " < " << diskOffsetLimit());
 715
 716     // Hopefully, we find a removable entry much sooner (TODO: use time?)
 717     const int maxProbed = 10000;
 718     RemovalPurgeWalker *walker = repl->PurgeInit(repl, maxProbed);
 719
 720     // It really should not take that long, but this will stop "infinite" loops
 721     const int maxFreed = 1000;
 722     int freed = 0;
 723     // TODO: should we purge more than needed to minimize overheads?
 724     for (; freed < maxFreed && full(); ++freed) {
 725         if (StoreEntry *e = walker->Next(walker))
 726             e->release(); // will call our unlink() method
 727         else
 728             break; // no more objects
 729     }
 730
 731     debugs(47,2, HERE << "Rock cache_dir[" << index << "] freed " << freed <<
 732            " scanned " << walker->scanned << '/' << walker->locked);
 733
 734     walker->Done(walker);
 735
 736     if (full()) {
 737         debugs(47, DBG_CRITICAL, "ERROR: Rock cache_dir[" << index << "] " <<
 738                "is still full after freeing " << freed << " entries. A bug?");
 739     }
 740 }
 741
 742 void
 743 Rock::SwapDir::reference(StoreEntry &e)
 744 {
 745     debugs(47, 5, HERE << &e << ' ' << e.swap_dirn << ' ' << e.swap_filen);
 746     if (repl && repl->Referenced)
 747         repl->Referenced(repl, &e, &e.repl);
 748 }
 749
 750 bool
 751 Rock::SwapDir::dereference(StoreEntry &e)
 752 {
 753     debugs(47, 5, HERE << &e << ' ' << e.swap_dirn << ' ' << e.swap_filen);
 754     if (repl && repl->Dereferenced)
 755         repl->Dereferenced(repl, &e, &e.repl);
 756
 757     // no need to keep e in the global store_table for us; we have our own map
 758     return false;
 759 }
 760
 761 bool
 762 Rock::SwapDir::unlinkdUseful() const
 763 {
 764     // no entry-specific files to unlink
 765     return false;
 766 }
 767
 768 void
 769 Rock::SwapDir::unlink(StoreEntry &e)
 770 {
 771     debugs(47, 5, HERE << e);
 772     ignoreReferences(e);
 773     map->free(e.swap_filen);
 774     disconnect(e);
 775 }
 776
 777 void
 778 Rock::SwapDir::trackReferences(StoreEntry &e)
 779 {
 780     debugs(47, 5, HERE << e);
 781     if (repl)
 782         repl->Add(repl, &e, &e.repl);
 783 }
 784
 785 void
 786 Rock::SwapDir::ignoreReferences(StoreEntry &e)
 787 {
 788     debugs(47, 5, HERE << e);
 789     if (repl)
 790         repl->Remove(repl, &e, &e.repl);
 791 }
 792
 793 void
 794 Rock::SwapDir::statfs(StoreEntry &e) const
 795 {
 796     storeAppendPrintf(&e, "\n");
 797     storeAppendPrintf(&e, "Maximum Size: %" PRIu64 " KB\n", maxSize() >> 10);
 798     storeAppendPrintf(&e, "Current Size: %.2f KB %.2f%%\n",
 799                       currentSize() / 1024.0,
 800                       Math::doublePercent(currentSize(), maxSize()));
 801
 802     if (map) {
 803         const int limit = map->entryLimit();
 804         storeAppendPrintf(&e, "Maximum entries: %9d\n", limit);
 805         if (limit > 0) {
 806             const int entryCount = map->entryCount();
 807             storeAppendPrintf(&e, "Current entries: %9d %.2f%%\n",
 808                               entryCount, (100.0 * entryCount / limit));
 809
 810             if (limit < 100) { // XXX: otherwise too expensive to count
 811                 Ipc::ReadWriteLockStats stats;
 812                 map->updateStats(stats);
 813                 stats.dump(e);
 814             }
 815         }
 816     }
 817
 818     storeAppendPrintf(&e, "Pending operations: %d out of %d\n",
 819                       store_open_disk_fd, Config.max_open_disk_fds);
 820
 821     storeAppendPrintf(&e, "Flags:");
 822
 823     if (flags.selected)
 824         storeAppendPrintf(&e, " SELECTED");
 825
 826     if (flags.read_only)
 827         storeAppendPrintf(&e, " READ-ONLY");
 828
 829     storeAppendPrintf(&e, "\n");
 830
 831 }
 832
 833 namespace Rock
 834 {
 835 RunnerRegistrationEntry(rrAfterConfig, SwapDirRr);
 836 }
 837
 838 void Rock::SwapDirRr::create(const RunnerRegistry &)
 839 {
 840     Must(owners.empty());
 841     for (int i = 0; i < Config.cacheSwap.n_configured; ++i) {
 842         if (const Rock::SwapDir *const sd = dynamic_cast<Rock::SwapDir *>(INDEXSD(i))) {
 843             Rock::SwapDir::DirMap::Owner *const owner =
 844                 Rock::SwapDir::DirMap::Init(sd->path, sd->entryLimitAllowed());
 845             owners.push_back(owner);
 846         }
 847     }
 848 }
 849
 850 Rock::SwapDirRr::~SwapDirRr()
 851 {
 852     for (size_t i = 0; i < owners.size(); ++i)
 853         delete owners[i];
 854 }