src/comm/Write.cc

   1 #include "squid.h"
   2 #include "comm/Connection.h"
   3 #include "comm/IoCallback.h"
   4 #include "comm/Write.h"
   5 #include "fd.h"
   6 #include "fde.h"
   7 #include "globals.h"
   8 #include "MemBuf.h"
   9 #include "profiler/Profiler.h"
  10 #include "SquidTime.h"
  11 #include "StatCounters.h"
  12
  13 #if USE_DELAY_POOLS
  14 #include "ClientInfo.h"
  15 #endif
  16 #if HAVE_ERRNO_H
  17 #include <errno.h>
  18 #endif
  19
  20 void
  21 Comm::Write(const Comm::ConnectionPointer &conn, MemBuf *mb, AsyncCall::Pointer &callback)
  22 {
  23     Comm::Write(conn, mb->buf, mb->size, callback, mb->freeFunc());
  24 }
  25
  26 void
  27 Comm::Write(const Comm::ConnectionPointer &conn, const char *buf, int size, AsyncCall::Pointer &callback, FREE * free_func)
  28 {
  29     debugs(5, 5, HERE << conn << ": sz " << size << ": asynCall " << callback);
  30
  31     /* Make sure we are open, not closing, and not writing */
  32     assert(fd_table[conn->fd].flags.open);
  33     assert(!fd_table[conn->fd].closing());
  34     Comm::IoCallback *ccb = COMMIO_FD_WRITECB(conn->fd);
  35     assert(!ccb->active());
  36
  37     fd_table[conn->fd].writeStart = squid_curtime;
  38     ccb->conn = conn;
  39     /* Queue the write */
  40     ccb->setCallback(IOCB_WRITE, callback, (char *)buf, free_func, size);
  41     ccb->selectOrQueueWrite();
  42 }
  43
  44 /** Write to FD.
  45  * This function is used by the lowest level of IO loop which only has access to FD numbers.
  46  * We have to use the comm iocb_table to map FD numbers to waiting data and Comm::Connections.
  47  * Once the write has been concluded we schedule the waiting call with success/fail results.
  48  */
  49 void
  50 Comm::HandleWrite(int fd, void *data)
  51 {
  52     Comm::IoCallback *state = static_cast<Comm::IoCallback *>(data);
  53     int len = 0;
  54     int nleft;
  55
  56     assert(state->conn != NULL && state->conn->fd == fd);
  57
  58     PROF_start(commHandleWrite);
  59     debugs(5, 5, HERE << state->conn << ": off " <<
  60            (long int) state->offset << ", sz " << (long int) state->size << ".");
  61
  62     nleft = state->size - state->offset;
  63
  64 #if USE_DELAY_POOLS
  65     ClientInfo * clientInfo=fd_table[fd].clientInfo;
  66
  67     if (clientInfo && !clientInfo->writeLimitingActive)
  68         clientInfo = NULL; // we only care about quota limits here
  69
  70     if (clientInfo) {
  71         assert(clientInfo->selectWaiting);
  72         clientInfo->selectWaiting = false;
  73
  74         assert(clientInfo->hasQueue());
  75         assert(clientInfo->quotaPeekFd() == fd);
  76         clientInfo->quotaDequeue(); // we will write or requeue below
  77
  78         if (nleft > 0) {
  79             const int quota = clientInfo->quotaForDequed();
  80             if (!quota) {  // if no write quota left, queue this fd
  81                 state->quotaQueueReserv = clientInfo->quotaEnqueue(fd);
  82                 clientInfo->kickQuotaQueue();
  83                 PROF_stop(commHandleWrite);
  84                 return;
  85             }
  86
  87             const int nleft_corrected = min(nleft, quota);
  88             if (nleft != nleft_corrected) {
  89                 debugs(5, 5, HERE << state->conn << " writes only " <<
  90                        nleft_corrected << " out of " << nleft);
  91                 nleft = nleft_corrected;
  92             }
  93
  94         }
  95     }
  96 #endif /* USE_DELAY_POOLS */
  97
  98     /* actually WRITE data */
  99     len = FD_WRITE_METHOD(fd, state->buf + state->offset, nleft);
 100     debugs(5, 5, HERE << "write() returns " << len);
 101
 102 #if USE_DELAY_POOLS
 103     if (clientInfo) {
 104         if (len > 0) {
 105             /* we wrote data - drain them from bucket */
 106             clientInfo->bucketSize -= len;
 107             if (clientInfo->bucketSize < 0.0) {
 108                 debugs(5, DBG_IMPORTANT, HERE << "drained too much"); // should not happen
 109                 clientInfo->bucketSize = 0;
 110             }
 111         }
 112
 113         // even if we wrote nothing, we were served; give others a chance
 114         clientInfo->kickQuotaQueue();
 115     }
 116 #endif /* USE_DELAY_POOLS */
 117
 118     fd_bytes(fd, len, FD_WRITE);
 119     ++statCounter.syscalls.sock.writes;
 120     // After each successful partial write,
 121     // reset fde::writeStart to the current time.
 122     fd_table[fd].writeStart = squid_curtime;
 123
 124     if (len == 0) {
 125         /* Note we even call write if nleft == 0 */
 126         /* We're done */
 127         if (nleft != 0)
 128             debugs(5, DBG_IMPORTANT, "FD " << fd << " write failure: connection closed with " << nleft << " bytes remaining.");
 129
 130         state->finish(nleft ? COMM_ERROR : COMM_OK, errno);
 131     } else if (len < 0) {
 132         /* An error */
 133         if (fd_table[fd].flags.socket_eof) {
 134             debugs(50, 2, HERE << "FD " << fd << " write failure: " << xstrerror() << ".");
 135             state->finish(nleft ? COMM_ERROR : COMM_OK, errno);
 136         } else if (ignoreErrno(errno)) {
 137             debugs(50, 9, HERE << "FD " << fd << " write failure: " << xstrerror() << ".");
 138             state->selectOrQueueWrite();
 139         } else {
 140             debugs(50, 2, HERE << "FD " << fd << " write failure: " << xstrerror() << ".");
 141             state->finish(nleft ? COMM_ERROR : COMM_OK, errno);
 142         }
 143     } else {
 144         /* A successful write, continue */
 145         state->offset += len;
 146
 147         if (state->offset < state->size) {
 148             /* Not done, reinstall the write handler and write some more */
 149             state->selectOrQueueWrite();
 150         } else {
 151             state->finish(nleft ? COMM_OK : COMM_ERROR, errno);
 152         }
 153     }
 154
 155     PROF_stop(commHandleWrite);
 156 }