src/fs/rock/RockIoState.cc

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