src/comm/IoCallback.cc

   1 #include "config.h"
   2 #include "ClientInfo.h"
   3 #include "comm/IoCallback.h"
   4 #include "comm/Write.h"
   5 #include "CommCalls.h"
   6 #include "fde.h"
   7
   8 Comm::CbEntry *Comm::iocb_table;
   9
  10 void
  11 Comm::CallbackTableInit()
  12 {
  13     // XXX: convert this to a std::map<> ?
  14     iocb_table = static_cast<CbEntry*>(xcalloc(Squid_MaxFD, sizeof(CbEntry)));
  15     for (int pos = 0; pos < Squid_MaxFD; pos++) {
  16         iocb_table[pos].fd = pos;
  17         iocb_table[pos].readcb.fd = pos;
  18         iocb_table[pos].readcb.type = IOCB_READ;
  19         iocb_table[pos].writecb.fd = pos;
  20         iocb_table[pos].writecb.type = IOCB_WRITE;
  21     }
  22 }
  23
  24 void
  25 Comm::CallbackTableDestruct()
  26 {
  27     safe_free(iocb_table);
  28 }
  29
  30 /**
  31  * Configure Comm::Callback for I/O
  32  *
  33  * @param fd            filedescriptor
  34  * @param t             IO callback type (read or write)
  35  * @param cb            callback
  36  * @param buf           buffer, if applicable
  37  * @param func          freefunc, if applicable
  38  * @param sz            buffer size
  39  */
  40 void
  41 Comm::IoCallback::setCallback(Comm::iocb_type t, AsyncCall::Pointer &cb, char *b, FREE *f, int sz)
  42 {
  43     assert(!active());
  44     assert(type == t);
  45     assert(cb != NULL);
  46
  47     callback = cb;
  48     buf = b;
  49     freefunc = f;
  50     size = sz;
  51     offset = 0;
  52 }
  53
  54 void
  55 Comm::IoCallback::selectOrQueueWrite()
  56 {
  57 #if USE_DELAY_POOLS
  58     // stand in line if there is one
  59     if (ClientInfo *clientInfo = fd_table[fd].clientInfo) {
  60         if (clientInfo->writeLimitingActive) {
  61             quotaQueueReserv = clientInfo->quotaEnqueue(fd);
  62             clientInfo->kickQuotaQueue();
  63             return;
  64         }
  65     }
  66 #endif
  67
  68     commSetSelect(fd, COMM_SELECT_WRITE, Comm::HandleWrite, this, 0);
  69 }
  70
  71 void
  72 Comm::IoCallback::cancel(const char *reason)
  73 {
  74     if (!active())
  75         return;
  76
  77     callback->cancel(reason);
  78     callback = NULL;
  79     reset();
  80 }
  81
  82 void
  83 Comm::IoCallback::reset()
  84 {
  85     if (freefunc) {
  86         freefunc(buf);
  87         buf = NULL;
  88         freefunc = NULL;
  89     }
  90     xerrno = 0;
  91
  92 #if USE_DELAY_POOLS
  93     quotaQueueReserv = 0;
  94 #endif
  95 }
  96
  97 // Schedule the callback call and clear the callback
  98 void
  99 Comm::IoCallback::finish(comm_err_t code, int xerrn)
 100 {
 101     debugs(5, 3, HERE << "called for FD " << fd << " (" << code << ", " << xerrno << ")");
 102     assert(active());
 103
 104     /* free data */
 105     if (freefunc) {
 106         freefunc(buf);
 107         buf = NULL;
 108         freefunc = NULL;
 109     }
 110
 111     if (callback != NULL) {
 112         typedef CommIoCbParams Params;
 113         Params &params = GetCommParams<Params>(callback);
 114         params.fd = fd;
 115         params.buf = buf;
 116         params.size = offset;
 117         params.flag = code;
 118         params.xerrno = xerrn;
 119         ScheduleCallHere(callback);
 120         callback = NULL;
 121     }
 122
 123     /* Reset for next round. */
 124     reset();
 125 }