Roll Comm::Connection into delayed reads

diff --git a/src/CommCalls.cc b/src/CommCalls.cc
index 4191fed73e50e6629386ae96d3033047daf70833..b656b810405779c4710bc9fca3d8ed59ef70c443 100644 (file)
--- a/src/CommCalls.cc
+++ b/src/CommCalls.cc
@@ -6,12 +6,12 @@
 /* CommCommonCbParams */
 
 CommCommonCbParams::CommCommonCbParams(void *aData):
-        data(cbdataReference(aData)), fd(-1), xerrno(0), flag(COMM_OK)
+        data(cbdataReference(aData)), conn(), flag(COMM_OK), xerrno(0), fd(-1)
 {
 }
 
 CommCommonCbParams::CommCommonCbParams(const CommCommonCbParams &p):
-        data(cbdataReference(p.data)), fd(p.fd), xerrno(p.xerrno), flag(p.flag)
+        data(cbdataReference(p.data)), conn(p.conn), flag(p.flag), xerrno(p.xerrno), fd(p.fd)
 {
 }
 

diff --git a/src/CommCalls.h b/src/CommCalls.h
index 00785e7254ff3962379034513c4449e1132177e5..cb8a9fc540a3fadd6a4073a7058b2691ff0dc454 100644 (file)
--- a/src/CommCalls.h
+++ b/src/CommCalls.h
@@ -54,11 +54,22 @@ public:
 
 public:
     void *data; // cbdata-protected
+
+    /** The connection which this call pertains to.
+     * \itemize On accept() calls this is the new client connection.
+     * \itemize On connect() finished calls this is the newely opened connection.
+     * \itemize On write calls this is the connection just written to.
+     * \itemize On read calls this is the connection just read from.
+     * \itemize On close calls this describes the connection which is now closed.
+     * \itemize On timeouts this is the connection whose operation timed out.
+     *          NP: timeouts should now return to the connect/read/write handler with COMM_ERR_TIMEOUT.
+     */
     Comm::ConnectionPointer conn;
-    int fd; // raw FD from legacy calls. use conn instead.
-    int xerrno;
-    comm_err_t flag;
 
+    comm_err_t flag;  ///< comm layer result status.
+    int xerrno;      ///< The last errno to occur. non-zero if flag is COMM_ERR.
+
+    int fd; // raw FD which the call was about. use conn instead for new code.
 private:
     // should not be needed and not yet implemented
     CommCommonCbParams &operator =(const CommCommonCbParams &params);

diff --git a/src/CommRead.h b/src/CommRead.h
index 1c0f6fc3356401d49f92f98fe9782d18b09f7424..da23379254d32128741d3a7d0a8646642c5f1f43 100644 (file)
--- a/src/CommRead.h
+++ b/src/CommRead.h
@@ -39,15 +39,16 @@
 #include "squid.h"
 #include "comm.h"
 #include "CommCalls.h"
+#include "comm/forward.h"
 #include "CbDataList.h"
 
 class CommRead
 {
 
 public:
-    CommRead ();
-    CommRead (int fd, char *buf, int len, AsyncCall::Pointer &callback);
-    int fd;
+    CommRead();
+    CommRead(const Comm::ConnectionPointer &c, char *buf, int len, AsyncCall::Pointer &callback);
+    Comm::ConnectionPointer conn;
     char *buf;
     int len;
     AsyncCall::Pointer callback;

diff --git a/src/MemObject.cc b/src/MemObject.cc
index 1f93fe93cbaa8460c004357557519eb4667c0525..472adca78eca4b831737d525246326aa753a3d6f 100644 (file)
--- a/src/MemObject.cc
+++ b/src/MemObject.cc
@@ -34,6 +34,7 @@
  */
 
 #include "squid.h"
+#include "comm/Connection.h"
 #include "MemObject.h"
 #include "HttpRequest.h"
 #include "HttpReply.h"

diff --git a/src/Store.h b/src/Store.h
index af0f6b3e7713bb78161a85ca85167c7a31a595f4..b89acba25570851a4fde7545abd994c409eac0bc 100644 (file)
--- a/src/Store.h
+++ b/src/Store.h
@@ -42,6 +42,7 @@
 #include "Range.h"
 #include "RefCount.h"
 #include "CommRead.h"
+#include "comm/forward.h"
 #include "Packer.h"
 #include "RemovalPolicy.h"
 
@@ -131,7 +132,7 @@ public:
     void destroyMemObject();
     int checkTooSmall();
 
-    void delayAwareRead(int fd, char *buf, int len, AsyncCall::Pointer callback);
+    void delayAwareRead(const Comm::ConnectionPointer &conn, char *buf, int len, AsyncCall::Pointer callback);
 
     void setNoDelay (bool const);
     bool modifiedSince(HttpRequest * request) const;

diff --git a/src/comm.cc b/src/comm.cc
index d71dc7af7770ae9f764080c7dc4a09992054ee1b..a4ff2ef1eb570f308fdab738eb4ff3935f38c98e 100644 (file)
--- a/src/comm.cc
+++ b/src/comm.cc
@@ -2021,10 +2021,10 @@ commHalfClosedReader(const Comm::ConnectionPointer &conn, char *, size_t size, c
 }
 
 
-CommRead::CommRead() : fd(-1), buf(NULL), len(0), callback(NULL) {}
+CommRead::CommRead() : conn(NULL), buf(NULL), len(0), callback(NULL) {}
 
-CommRead::CommRead(int fd_, char *buf_, int len_, AsyncCall::Pointer &callback_)
-        : fd(fd_), buf(buf_), len(len_), callback(callback_) {}
+CommRead::CommRead(const Comm::ConnectionPointer &c, char *buf_, int len_, AsyncCall::Pointer &callback_)
+        : conn(c), buf(buf_), len(len_), callback(callback_) {}
 
 DeferredRead::DeferredRead () : theReader(NULL), theContext(NULL), theRead(), cancelled(false) {}
 
@@ -2045,7 +2045,7 @@ template cbdata_type CbDataList<DeferredRead>::CBDATA_CbDataList;
 void
 DeferredReadManager::delayRead(DeferredRead const &aRead)
 {
-    debugs(5, 3, "Adding deferred read on FD " << aRead.theRead.fd);
+    debugs(5, 3, "Adding deferred read on " << aRead.theRead.conn);
     CbDataList<DeferredRead> *temp = deferredReads.push_back(aRead);
 
     // We have to use a global function as a closer and point to temp
@@ -2054,7 +2054,7 @@ DeferredReadManager::delayRead(DeferredRead const &aRead)
     AsyncCall::Pointer closer = commCbCall(5,4,
                                            "DeferredReadManager::CloseHandler",
                                            CommCloseCbPtrFun(&CloseHandler, temp));
-    comm_add_close_handler(aRead.theRead.fd, closer);
+    comm_add_close_handler(aRead.theRead.conn->fd, closer);
     temp->element.closer = closer; // remeber so that we can cancel
 }
 
@@ -2077,7 +2077,7 @@ DeferredReadManager::popHead(CbDataListContainer<DeferredRead> &deferredReads)
 
     DeferredRead &read = deferredReads.head->element;
     if (!read.cancelled) {
-        comm_remove_close_handler(read.theRead.fd, read.closer);
+        comm_remove_close_handler(read.theRead.conn->fd, read.closer);
         read.closer = NULL;
     }
 
@@ -2127,10 +2127,10 @@ DeferredReadManager::kickARead(DeferredRead const &aRead)
     if (aRead.cancelled)
         return;
 
-    if (aRead.theRead.fd>=0 && fd_table[aRead.theRead.fd].closing())
+    if (Comm::IsConnOpen(aRead.theRead.conn) && fd_table[aRead.theRead.conn->fd].closing())
         return;
 
-    debugs(5, 3, "Kicking deferred read on FD " << aRead.theRead.fd);
+    debugs(5, 3, "Kicking deferred read on " << aRead.theRead.conn);
 
     aRead.theReader(aRead.theContext, aRead.theRead);
 }

diff --git a/src/comm/Connection.cc b/src/comm/Connection.cc
index 766071020d35ba77d803d669d7332d5d19fed158..8a2a04b04ad102a726cfe5dc4563cd907b1b01bb 100644 (file)
--- a/src/comm/Connection.cc
+++ b/src/comm/Connection.cc
@@ -24,6 +24,8 @@ Comm::Connection::Connection() :
 static int64_t lost_conn = 0;
 Comm::Connection::~Connection()
 {
+    assert(fd < 0); // These should never occur now.
+
     if (fd >= 0) {
         debugs(5, 0, "NOTE: Orphan Comm::Connection: " << *this);
         debugs(5, 0, "NOTE: Orphaned Comm::Connections: " << ++lost_conn);

diff --git a/src/ftp.cc b/src/ftp.cc
index ecdd3caf2ba58c42ef4fec23ae6becf31e4a5628..6c2eb0af892c9c9e7c1bd4859a9dae10f8b8b660 100644 (file)
--- a/src/ftp.cc
+++ b/src/ftp.cc
@@ -1201,7 +1201,7 @@ FtpStateData::maybeReadVirginBody()
     debugs(9,5,HERE << "queueing read on FD " << data.conn->fd);
 
     typedef CommCbMemFunT<FtpStateData, CommIoCbParams> Dialer;
-    entry->delayAwareRead(data.conn->fd, data.readBuf->space(), read_sz,
+    entry->delayAwareRead(data.conn, data.readBuf->space(), read_sz,
                           JobCallback(9, 5, Dialer, this, FtpStateData::dataRead));
 }
 

diff --git a/src/gopher.cc b/src/gopher.cc
index 52620934c20bb419713cd98b46e9a3708b4644cf..b46045bab9d2a6474b903fcfcb6a6baf4764ed5b 100644 (file)
--- a/src/gopher.cc
+++ b/src/gopher.cc
@@ -940,7 +940,7 @@ gopherSendComplete(const Comm::ConnectionPointer &conn, char *buf, size_t size,
     /* Schedule read reply. */
     AsyncCall::Pointer call =  commCbCall(10,5, "gopherReadReply",
                                           CommIoCbPtrFun(gopherReadReply, gopherState));
-    entry->delayAwareRead(conn->fd, gopherState->replybuf, BUFSIZ, call);
+    entry->delayAwareRead(conn, gopherState->replybuf, BUFSIZ, call);
 
     if (buf)
         memFree(buf, MEM_4K_BUF);      /* Allocated by gopherSendRequest. */

diff --git a/src/http.cc b/src/http.cc
index dad0dee729fc627637690ec886d857e2772c1bc8..3de71c1ec32b055e34b8e8b1485e6dfd9c31e5ef 100644 (file)
--- a/src/http.cc
+++ b/src/http.cc
@@ -1088,7 +1088,7 @@ HttpStateData::readReply(const CommIoCbParams &io)
 
     flags.do_next_read = 0;
 
-    debugs(11, 5, "httpReadReply: FD " << io.fd << ": len " << len << ".");
+    debugs(11, 5, HERE << io.conn << ": len " << len << ".");
 
     // Bail out early on COMM_ERR_CLOSING - close handlers will tidy up for us
     if (io.flag == COMM_ERR_CLOSING) {
@@ -1103,7 +1103,7 @@ HttpStateData::readReply(const CommIoCbParams &io)
 
     // handle I/O errors
     if (io.flag != COMM_OK || len < 0) {
-        debugs(11, 2, "httpReadReply: FD " << io.fd << ": read failure: " << xstrerror() << ".");
+        debugs(11, 2, HERE << io.conn << ": read failure: " << xstrerror() << ".");
 
         if (ignoreErrno(io.xerrno)) {
             flags.do_next_read = 1;
@@ -1398,7 +1398,7 @@ HttpStateData::processReplyBody()
     } else
         switch (persistentConnStatus()) {
         case INCOMPLETE_MSG:
-            debugs(11, 5, "processReplyBody: INCOMPLETE_MSG");
+            debugs(11, 5, "processReplyBody: INCOMPLETE_MSG from " << serverConnection);
             /* Wait for more data or EOF condition */
             if (flags.keepalive_broken) {
                 call = NULL;
@@ -1412,7 +1412,7 @@ HttpStateData::processReplyBody()
             break;
 
         case COMPLETE_PERSISTENT_MSG:
-            debugs(11, 5, "processReplyBody: COMPLETE_PERSISTENT_MSG");
+            debugs(11, 5, "processReplyBody: COMPLETE_PERSISTENT_MSG from " << serverConnection);
             /* yes we have to clear all these! */
             call = NULL;
             commSetTimeout(serverConnection->fd, -1, call);
@@ -1440,12 +1440,11 @@ HttpStateData::processReplyBody()
             }
 
             serverConnection = NULL;
-
             serverComplete();
             return;
 
         case COMPLETE_NONPERSISTENT_MSG:
-            debugs(11, 5, "processReplyBody: COMPLETE_NONPERSISTENT_MSG");
+            debugs(11, 5, "processReplyBody: COMPLETE_NONPERSISTENT_MSG from " << serverConnection);
             serverComplete();
             return;
         }
@@ -1477,7 +1476,7 @@ HttpStateData::maybeReadVirginBody()
     if (flags.do_next_read) {
         flags.do_next_read = 0;
         typedef CommCbMemFunT<HttpStateData, CommIoCbParams> Dialer;
-        entry->delayAwareRead(serverConnection->fd, readBuf->space(read_size), read_size,
+        entry->delayAwareRead(serverConnection, readBuf->space(read_size), read_size,
                               JobCallback(11, 5, Dialer, this,  HttpStateData::readReply));
     }
 }

diff --git a/src/store.cc b/src/store.cc
index 499efd6a60043706ab1c864962d456e7d0db1561..f0e182bdd6b4bea6ed0d0e3e5db0cc54e94033f9 100644 (file)
--- a/src/store.cc
+++ b/src/store.cc
@@ -221,14 +221,14 @@ void
 StoreEntry::DeferReader(void *theContext, CommRead const &aRead)
 {
     StoreEntry *anEntry = (StoreEntry *)theContext;
-    anEntry->delayAwareRead(aRead.fd,
+    anEntry->delayAwareRead(aRead.conn,
                             aRead.buf,
                             aRead.len,
                             aRead.callback);
 }
 
 void
-StoreEntry::delayAwareRead(int fd, char *buf, int len, AsyncCall::Pointer callback)
+StoreEntry::delayAwareRead(const Comm::ConnectionPointer &conn, char *buf, int len, AsyncCall::Pointer callback)
 {
     size_t amountToRead = bytesWanted(Range<size_t>(0, len));
     /* sketch: readdeferer* = getdeferer.
@@ -242,23 +242,18 @@ StoreEntry::delayAwareRead(int fd, char *buf, int len, AsyncCall::Pointer callba
 #if USE_DELAY_POOLS
         if (!mem_obj->readAheadPolicyCanRead()) {
 #endif
-            mem_obj->delayRead(DeferredRead(DeferReader, this, CommRead(fd, buf, len, callback)));
+            mem_obj->delayRead(DeferredRead(DeferReader, this, CommRead(conn, buf, len, callback)));
             return;
 #if USE_DELAY_POOLS
         }
 
         /* delay id limit */
-        mem_obj->mostBytesAllowed().delayRead(DeferredRead(DeferReader, this, CommRead(fd, buf, len, callback)));
-
+        mem_obj->mostBytesAllowed().delayRead(DeferredRead(DeferReader, this, CommRead(conn, buf, len, callback)));
         return;
-
 #endif
-
     }
 
-    Comm::ConnectionPointer temp = new Comm::Connection; // XXX: transition. until conn passed in.
-    temp->fd = fd;
-    comm_read(temp, buf, amountToRead, callback);
+    comm_read(conn, buf, amountToRead, callback);
 }
 
 size_t