src/fs/rock/RockIoState.cc

   1 /*
   2  * DEBUG: section 79    Disk IO Routines
   3  */
   4
   5 #include "squid.h"
   6 #include "base/TextException.h"
   7 #include "CollapsedForwarding.h"
   8 #include "DiskIO/DiskIOModule.h"
   9 #include "DiskIO/DiskIOStrategy.h"
  10 #include "DiskIO/WriteRequest.h"
  11 #include "fs/rock/RockIoRequests.h"
  12 #include "fs/rock/RockIoState.h"
  13 #include "fs/rock/RockSwapDir.h"
  14 #include "globals.h"
  15 #include "Mem.h"
  16 #include "MemObject.h"
  17 #include "Parsing.h"
  18 #include "Transients.h"
  19
  20 Rock::IoState::IoState(Rock::SwapDir::Pointer &aDir,
  21                        StoreEntry *anEntry,
  22                        StoreIOState::STFNCB *cbFile,
  23                        StoreIOState::STIOCB *cbIo,
  24                        void *data):
  25         readableAnchor_(NULL),
  26         writeableAnchor_(NULL),
  27         sidCurrent(-1),
  28         dir(aDir),
  29         slotSize(dir->slotSize),
  30         objOffset(0),
  31         theBuf(dir->slotSize)
  32 {
  33     e = anEntry;
  34     e->lock("rock I/O");
  35     // anchor, swap_filen, and swap_dirn are set by the caller
  36     file_callback = cbFile;
  37     callback = cbIo;
  38     callback_data = cbdataReference(data);
  39     ++store_open_disk_fd; // TODO: use a dedicated counter?
  40     //theFile is set by SwapDir because it depends on DiskIOStrategy
  41 }
  42
  43 Rock::IoState::~IoState()
  44 {
  45     --store_open_disk_fd;
  46
  47     // The dir map entry may still be open for reading at the point because
  48     // the map entry lock is associated with StoreEntry, not IoState.
  49     // assert(!readableAnchor_);
  50     assert(shutting_down || !writeableAnchor_);
  51
  52     if (callback_data)
  53         cbdataReferenceDone(callback_data);
  54     theFile = NULL;
  55
  56     e->unlock("rock I/O");
  57 }
  58
  59 void
  60 Rock::IoState::file(const RefCount<DiskFile> &aFile)
  61 {
  62     assert(!theFile);
  63     assert(aFile != NULL);
  64     theFile = aFile;
  65 }
  66
  67 const Ipc::StoreMapAnchor &
  68 Rock::IoState::readAnchor() const
  69 {
  70     assert(readableAnchor_);
  71     return *readableAnchor_;
  72 }
  73
  74 Ipc::StoreMapAnchor &
  75 Rock::IoState::writeAnchor()
  76 {
  77     assert(writeableAnchor_);
  78     return *writeableAnchor_;
  79 }
  80
  81 /// convenience wrapper returning the map slot we are reading now
  82 const Ipc::StoreMapSlice &
  83 Rock::IoState::currentReadableSlice() const
  84 {
  85     return dir->map->readableSlice(swap_filen, sidCurrent);
  86 }
  87
  88 void
  89 Rock::IoState::read_(char *buf, size_t len, off_t coreOff, STRCB *cb, void *data)
  90 {
  91     debugs(79, 7, swap_filen << " reads from " << coreOff);
  92
  93     assert(theFile != NULL);
  94     assert(coreOff >= 0);
  95
  96     // if we are dealing with the first read or
  97     // if the offset went backwords, start searching from the beginning
  98     if (sidCurrent < 0 || coreOff < objOffset) {
  99         sidCurrent = readAnchor().start;
 100         objOffset = 0;
 101     }
 102
 103     while (sidCurrent >= 0 && coreOff >= objOffset + currentReadableSlice().size) {
 104         objOffset += currentReadableSlice().size;
 105         sidCurrent = currentReadableSlice().next;
 106     }
 107
 108     assert(read.callback == NULL);
 109     assert(read.callback_data == NULL);
 110     read.callback = cb;
 111     read.callback_data = cbdataReference(data);
 112
 113     // punt if read offset is too big (because of client bugs or collapsing)
 114     if (sidCurrent < 0) {
 115         debugs(79, 5, "no " << coreOff << " in " << *e);
 116         callReaderBack(buf, 0);
 117         return;
 118     }
 119
 120     offset_ = coreOff;
 121     len = min(len,
 122               static_cast<size_t>(objOffset + currentReadableSlice().size - coreOff));
 123     const uint64_t diskOffset = dir->diskOffset(sidCurrent);
 124     theFile->read(new ReadRequest(::ReadRequest(buf,
 125                                   diskOffset + sizeof(DbCellHeader) + coreOff - objOffset, len), this));
 126 }
 127
 128 void
 129 Rock::IoState::callReaderBack(const char *buf, int rlen)
 130 {
 131     debugs(79, 5, rlen << " bytes for " << *e);
 132     StoreIOState::STRCB *callb = read.callback;
 133     assert(callb);
 134     read.callback = NULL;
 135     void *cbdata;
 136     if (cbdataReferenceValidDone(read.callback_data, &cbdata))
 137         callb(cbdata, buf, rlen, this);
 138 }
 139
 140 /// wraps tryWrite() to handle deep write failures centrally and safely
 141 bool
 142 Rock::IoState::write(char const *buf, size_t size, off_t coreOff, FREE *dtor)
 143 {
 144     bool success = false;
 145     try {
 146         tryWrite(buf, size, coreOff);
 147         success = true;
 148     } catch (const std::exception &ex) { // TODO: should we catch ... as well?
 149         debugs(79, 2, "db write error: " << ex.what());
 150         dir->writeError(*e);
 151         finishedWriting(DISK_ERROR);
 152         // 'this' might be gone beyond this point; fall through to free buf
 153     }
 154
 155     // careful: 'this' might be gone here
 156
 157     if (dtor)
 158         (dtor)(const_cast<char*>(buf)); // cast due to a broken API?
 159
 160     return success;
 161 }
 162
 163 /**
 164  * Possibly send data to be written to disk:
 165  * We only write data when full slot is accumulated or when close() is called.
 166  * We buffer, in part, to avoid forcing OS to _read_ old unwritten portions of
 167  * the slot when the write does not end at the page or sector boundary.
 168  */
 169 void
 170 Rock::IoState::tryWrite(char const *buf, size_t size, off_t coreOff)
 171 {
 172     debugs(79, 7, swap_filen << " writes " << size << " more");
 173
 174     // either this is the first write or append; we do not support write gaps
 175     assert(!coreOff || coreOff == -1);
 176
 177     // allocate the first slice during the first write
 178     if (!coreOff) {
 179         assert(sidCurrent < 0);
 180         sidCurrent = reserveSlotForWriting(); // throws on failures
 181         assert(sidCurrent >= 0);
 182         writeAnchor().start = sidCurrent;
 183     }
 184
 185     // buffer incoming data in slot buffer and write overflowing or final slots
 186     // quit when no data left or we stopped writing on reentrant error
 187     while (size > 0 && theFile != NULL) {
 188         assert(sidCurrent >= 0);
 189         const size_t processed = writeToBuffer(buf, size);
 190         buf += processed;
 191         size -= processed;
 192         const bool overflow = size > 0;
 193
 194         // We do not write a full buffer without overflow because
 195         // we would not yet know what to set the nextSlot to.
 196         if (overflow) {
 197             const SlotId sidNext = reserveSlotForWriting(); // throws
 198             assert(sidNext >= 0);
 199             writeToDisk(sidNext);
 200         } else if (Store::Root().transientReaders(*e)) {
 201             // write partial buffer for all remote hit readers to see
 202             writeBufToDisk(-1, false);
 203         }
 204     }
 205
 206 }
 207
 208 /// Buffers incoming data for the current slot.
 209 /// \return the number of bytes buffered
 210 size_t
 211 Rock::IoState::writeToBuffer(char const *buf, size_t size)
 212 {
 213     // do not buffer a cell header for nothing
 214     if (!size)
 215         return 0;
 216
 217     if (!theBuf.size) {
 218         // will fill the header in writeToDisk when the next slot is known
 219         theBuf.appended(sizeof(DbCellHeader));
 220     }
 221
 222     size_t forCurrentSlot = min(size, static_cast<size_t>(theBuf.spaceSize()));
 223     theBuf.append(buf, forCurrentSlot);
 224     offset_ += forCurrentSlot; // so that Core thinks we wrote it
 225     return forCurrentSlot;
 226 }
 227
 228 /// write what was buffered during write() calls
 229 /// negative sidNext means this is the last write request for this entry
 230 void
 231 Rock::IoState::writeToDisk(const SlotId sidNext)
 232 {
 233     assert(theFile != NULL);
 234     assert(theBuf.size >= sizeof(DbCellHeader));
 235
 236     // TODO: if DiskIO module is mmap-based, we should be writing whole pages
 237     // to avoid triggering read-page;new_head+old_tail;write-page overheads
 238
 239     writeBufToDisk(sidNext, sidNext < 0);
 240     theBuf.clear();
 241
 242     sidCurrent = sidNext;
 243 }
 244
 245 /// creates and submits a request to write current slot buffer to disk
 246 /// eof is true if and only this is the last slot
 247 void
 248 Rock::IoState::writeBufToDisk(const SlotId sidNext, bool eof)
 249 {
 250     // no slots after the last/eof slot (but partial slots may have a nil next)
 251     assert(!eof || sidNext < 0);
 252
 253     // finalize db cell header
 254     DbCellHeader header;
 255     memcpy(header.key, e->key, sizeof(header.key));
 256     header.firstSlot = writeAnchor().start;
 257     header.nextSlot = sidNext;
 258     header.payloadSize = theBuf.size - sizeof(DbCellHeader);
 259     header.entrySize = eof ? offset_ : 0; // storeSwapOutFileClosed sets swap_file_sz after write
 260     header.version = writeAnchor().basics.timestamp;
 261
 262     // copy finalized db cell header into buffer
 263     memcpy(theBuf.mem, &header, sizeof(DbCellHeader));
 264
 265     // and now allocate another buffer for the WriteRequest so that
 266     // we can support concurrent WriteRequests (and to ease cleaning)
 267     // TODO: should we limit the number of outstanding requests?
 268     size_t wBufCap = 0;
 269     void *wBuf = memAllocBuf(theBuf.size, &wBufCap);
 270     memcpy(wBuf, theBuf.mem, theBuf.size);
 271
 272     const uint64_t diskOffset = dir->diskOffset(sidCurrent);
 273     debugs(79, 5, HERE << swap_filen << " at " << diskOffset << '+' <<
 274            theBuf.size);
 275
 276     WriteRequest *const r = new WriteRequest(
 277         ::WriteRequest(static_cast<char*>(wBuf), diskOffset, theBuf.size,
 278                        memFreeBufFunc(wBufCap)), this);
 279     r->sidCurrent = sidCurrent;
 280     r->sidNext = sidNext;
 281     r->eof = eof;
 282
 283     // theFile->write may call writeCompleted immediatelly
 284     theFile->write(r);
 285 }
 286
 287 /// finds and returns a free db slot to fill or throws
 288 Rock::SlotId
 289 Rock::IoState::reserveSlotForWriting()
 290 {
 291     Ipc::Mem::PageId pageId;
 292     if (dir->useFreeSlot(pageId))
 293         return pageId.number-1;
 294
 295     // This may happen when the number of available db slots is close to the
 296     // number of concurrent requests reading or writing those slots, which may
 297     // happen when the db is "small" compared to the request traffic OR when we
 298     // are rebuilding and have not loaded "many" entries or empty slots yet.
 299     throw TexcHere("ran out of free db slots");
 300 }
 301
 302 void
 303 Rock::IoState::finishedWriting(const int errFlag)
 304 {
 305     // we incremented offset_ while accumulating data in write()
 306     // we do not reset writeableAnchor_ here because we still keep the lock
 307     CollapsedForwarding::Broadcast(*e);
 308     callBack(errFlag);
 309 }
 310
 311 void
 312 Rock::IoState::close(int how)
 313 {
 314     debugs(79, 3, swap_filen << " offset: " << offset_ << " how: " << how <<
 315            " buf: " << theBuf.size << " callback: " << callback);
 316
 317     if (!theFile) {
 318         debugs(79, 3, "I/O already canceled");
 319         assert(!callback);
 320         // We keep writeableAnchor_ after callBack() on I/O errors.
 321         assert(!readableAnchor_);
 322         return;
 323     }
 324
 325     switch (how) {
 326     case wroteAll:
 327         assert(theBuf.size > 0); // we never flush last bytes on our own
 328         writeToDisk(-1); // flush last, yet unwritten slot to disk
 329         return; // writeCompleted() will callBack()
 330
 331     case writerGone:
 332         assert(writeableAnchor_);
 333         dir->writeError(*e); // abort a partially stored entry
 334         finishedWriting(DISK_ERROR);
 335         return;
 336
 337     case readerDone:
 338         callBack(0);
 339         return;
 340     }
 341 }
 342
 343 /// close callback (STIOCB) dialer: breaks dependencies and
 344 /// counts IOState concurrency level
 345 class StoreIOStateCb: public CallDialer
 346 {
 347 public:
 348     StoreIOStateCb(StoreIOState::STIOCB *cb, void *data, int err, const Rock::IoState::Pointer &anSio):
 349             callback(NULL),
 350             callback_data(NULL),
 351             errflag(err),
 352             sio(anSio) {
 353
 354         callback = cb;
 355         callback_data = cbdataReference(data);
 356     }
 357
 358     StoreIOStateCb(const StoreIOStateCb &cb):
 359             callback(NULL),
 360             callback_data(NULL),
 361             errflag(cb.errflag),
 362             sio(cb.sio) {
 363
 364         callback = cb.callback;
 365         callback_data = cbdataReference(cb.callback_data);
 366     }
 367
 368     virtual ~StoreIOStateCb() {
 369         cbdataReferenceDone(callback_data); // may be nil already
 370     }
 371
 372     void dial(AsyncCall &call) {
 373         void *cbd;
 374         if (cbdataReferenceValidDone(callback_data, &cbd) && callback)
 375             callback(cbd, errflag, sio.getRaw());
 376     }
 377
 378     bool canDial(AsyncCall &call) const {
 379         return cbdataReferenceValid(callback_data) && callback;
 380     }
 381
 382     virtual void print(std::ostream &os) const {
 383         os << '(' << callback_data << ", err=" << errflag << ')';
 384     }
 385
 386 private:
 387     StoreIOStateCb &operator =(const StoreIOStateCb &cb); // not defined
 388
 389     StoreIOState::STIOCB *callback;
 390     void *callback_data;
 391     int errflag;
 392     Rock::IoState::Pointer sio;
 393 };
 394
 395 void
 396 Rock::IoState::callBack(int errflag)
 397 {
 398     debugs(79,3, HERE << "errflag=" << errflag);
 399     theFile = NULL;
 400
 401     AsyncCall::Pointer call = asyncCall(79,3, "SomeIoStateCloseCb",
 402                                         StoreIOStateCb(callback, callback_data, errflag, this));
 403     ScheduleCallHere(call);
 404
 405     callback = NULL;
 406     cbdataReferenceDone(callback_data);
 407 }
 408