src/fs/rock/RockSwapDir.cc

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