src/fs/rock/RockIoState.cc

   1 /*
   2  * DEBUG: section 79    Disk IO Routines
   3  */
   4
   5 #include "squid.h"
   6 #include "MemObject.h"
   7 #include "Parsing.h"
   8 #include "DiskIO/DiskIOModule.h"
   9 #include "DiskIO/DiskIOStrategy.h"
  10 #include "DiskIO/WriteRequest.h"
  11 #include "fs/rock/RockIoState.h"
  12 #include "fs/rock/RockIoRequests.h"
  13 #include "fs/rock/RockSwapDir.h"
  14 #include "globals.h"
  15
  16 Rock::IoState::IoState(SwapDir *dir,
  17                        StoreEntry *anEntry,
  18                        StoreIOState::STFNCB *cbFile,
  19                        StoreIOState::STIOCB *cbIo,
  20                        void *data):
  21         slotSize(0),
  22         diskOffset(-1),
  23         payloadEnd(-1)
  24 {
  25     e = anEntry;
  26     // swap_filen, swap_dirn, diskOffset, and payloadEnd are set by the caller
  27     slotSize = dir->maxObjectSize();
  28     file_callback = cbFile;
  29     callback = cbIo;
  30     callback_data = cbdataReference(data);
  31     ++store_open_disk_fd; // TODO: use a dedicated counter?
  32     //theFile is set by SwapDir because it depends on DiskIOStrategy
  33 }
  34
  35 Rock::IoState::~IoState()
  36 {
  37     --store_open_disk_fd;
  38     if (callback_data)
  39         cbdataReferenceDone(callback_data);
  40     theFile = NULL;
  41 }
  42
  43 void
  44 Rock::IoState::file(const RefCount<DiskFile> &aFile)
  45 {
  46     assert(!theFile);
  47     assert(aFile != NULL);
  48     theFile = aFile;
  49 }
  50
  51 void
  52 Rock::IoState::read_(char *buf, size_t len, off_t coreOff, STRCB *cb, void *data)
  53 {
  54     assert(theFile != NULL);
  55     assert(coreOff >= 0);
  56     offset_ = coreOff;
  57
  58     // we skip our cell header; it is only read when building the map
  59     const int64_t cellOffset = sizeof(DbCellHeader) +
  60                                static_cast<int64_t>(coreOff);
  61     assert(cellOffset <= payloadEnd);
  62
  63     // Core specifies buffer length, but we must not exceed stored entry size
  64     if (cellOffset + (int64_t)len > payloadEnd)
  65         len = payloadEnd - cellOffset;
  66
  67     assert(read.callback == NULL);
  68     assert(read.callback_data == NULL);
  69     read.callback = cb;
  70     read.callback_data = cbdataReference(data);
  71
  72     theFile->read(new ReadRequest(
  73                       ::ReadRequest(buf, diskOffset + cellOffset, len), this));
  74 }
  75
  76 // We only buffer data here; we actually write when close() is called.
  77 // We buffer, in part, to avoid forcing OS to _read_ old unwritten portions
  78 // of the slot when the write does not end at the page or sector boundary.
  79 void
  80 Rock::IoState::write(char const *buf, size_t size, off_t coreOff, FREE *dtor)
  81 {
  82     // TODO: move to create?
  83     if (!coreOff) {
  84         assert(theBuf.isNull());
  85         assert(payloadEnd <= slotSize);
  86         theBuf.init(min(payloadEnd, slotSize), slotSize);
  87         // start with our header; TODO: consider making it a trailer
  88         DbCellHeader header;
  89         assert(static_cast<int64_t>(sizeof(header)) <= payloadEnd);
  90         header.payloadSize = payloadEnd - sizeof(header);
  91         theBuf.append(reinterpret_cast<const char*>(&header), sizeof(header));
  92     } else {
  93         // Core uses -1 offset as "append". Sigh.
  94         assert(coreOff == -1);
  95         assert(!theBuf.isNull());
  96     }
  97
  98     theBuf.append(buf, size);
  99     offset_ += size; // so that Core thinks we wrote it
 100
 101     if (dtor)
 102         (dtor)(const_cast<char*>(buf)); // cast due to a broken API?
 103 }
 104
 105 // write what was buffered during write() calls
 106 void
 107 Rock::IoState::startWriting()
 108 {
 109     assert(theFile != NULL);
 110     assert(!theBuf.isNull());
 111
 112     // TODO: if DiskIO module is mmap-based, we should be writing whole pages
 113     // to avoid triggering read-page;new_head+old_tail;write-page overheads
 114
 115     debugs(79, 5, HERE << swap_filen << " at " << diskOffset << '+' <<
 116            theBuf.contentSize());
 117
 118     assert(theBuf.contentSize() <= slotSize);
 119     // theFile->write may call writeCompleted immediatelly
 120     theFile->write(new WriteRequest(::WriteRequest(theBuf.content(),
 121                                     diskOffset, theBuf.contentSize(), theBuf.freeFunc()), this));
 122 }
 123
 124 //
 125 void
 126 Rock::IoState::finishedWriting(const int errFlag)
 127 {
 128     // we incremented offset_ while accumulating data in write()
 129     callBack(errFlag);
 130 }
 131
 132 void
 133 Rock::IoState::close(int how)
 134 {
 135     debugs(79, 3, HERE << swap_filen << " accumulated: " << offset_ <<
 136            " how=" << how);
 137     if (how == wroteAll && !theBuf.isNull())
 138         startWriting();
 139     else
 140         callBack(how == writerGone ? DISK_ERROR : 0); // TODO: add DISK_CALLER_GONE
 141 }
 142
 143 /// close callback (STIOCB) dialer: breaks dependencies and
 144 /// counts IOState concurrency level
 145 class StoreIOStateCb: public CallDialer
 146 {
 147 public:
 148     StoreIOStateCb(StoreIOState::STIOCB *cb, void *data, int err, const Rock::IoState::Pointer &anSio):
 149             callback(NULL),
 150             callback_data(NULL),
 151             errflag(err),
 152             sio(anSio) {
 153
 154         callback = cb;
 155         callback_data = cbdataReference(data);
 156     }
 157
 158     StoreIOStateCb(const StoreIOStateCb &cb):
 159             callback(NULL),
 160             callback_data(NULL),
 161             errflag(cb.errflag),
 162             sio(cb.sio) {
 163
 164         callback = cb.callback;
 165         callback_data = cbdataReference(cb.callback_data);
 166     }
 167
 168     virtual ~StoreIOStateCb() {
 169         cbdataReferenceDone(callback_data); // may be nil already
 170     }
 171
 172     void dial(AsyncCall &call) {
 173         void *cbd;
 174         if (cbdataReferenceValidDone(callback_data, &cbd) && callback)
 175             callback(cbd, errflag, sio.getRaw());
 176     }
 177
 178     bool canDial(AsyncCall &call) const {
 179         return cbdataReferenceValid(callback_data) && callback;
 180     }
 181
 182     virtual void print(std::ostream &os) const {
 183         os << '(' << callback_data << ", err=" << errflag << ')';
 184     }
 185
 186 private:
 187     StoreIOStateCb &operator =(const StoreIOStateCb &cb); // not defined
 188
 189     StoreIOState::STIOCB *callback;
 190     void *callback_data;
 191     int errflag;
 192     Rock::IoState::Pointer sio;
 193 };
 194
 195 void
 196 Rock::IoState::callBack(int errflag)
 197 {
 198     debugs(79,3, HERE << "errflag=" << errflag);
 199     theFile = NULL;
 200
 201     AsyncCall::Pointer call = asyncCall(79,3, "SomeIoStateCloseCb",
 202                                         StoreIOStateCb(callback, callback_data, errflag, this));
 203     ScheduleCallHere(call);
 204
 205     callback = NULL;
 206     cbdataReferenceDone(callback_data);
 207 }
 208