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