src/comm/IoCallback.cc

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