Push SBuf down into Comm::IoCallback

Use a pointer to the SBuf instead of the rawSpace() because using a SBuf
copy can potentially cause different MemBlob to exist behind the Comm
read classes and the caller (ConnStateData) class. They need to be kept
identical so as not to loose any existing bytes in the I/O buffer when
a read callback handler is run.

Pointer to SBuf also avoids race conditions between comm_read async calls
and BodyPipe notifications which can potentially change the MemBlob
underneath comm_read() and invalidate te char* buffer pointer used
previously.

diff --git a/src/CommCalls.h b/src/CommCalls.h
index 084171b9281f0610b66a1695a25a7f4de9e2dcc3..8d3a9bac11de83ddac0342483c0f8d0a94d9e4a0 100644 (file)
--- a/src/CommCalls.h
+++ b/src/CommCalls.h
@@ -6,6 +6,7 @@
 #include "comm/forward.h"
 #include "comm_err_t.h"
 #include "MasterXaction.h"
+#include "SBuf.h"
 
 /* CommCalls implement AsyncCall interface for comm_* callbacks.
  * The classes cover two call dialer kinds:
@@ -118,6 +119,7 @@ public:
 public:
     char *buf;
     size_t size;
+    SBuf *buf2;  // alternative buffer for use when buf is unset
 };
 
 // close parameters

diff --git a/src/client_side.cc b/src/client_side.cc
index 54f11b455149306a4f36bc215bf1d823f6d86fab..87705cb4c706660c3edcfeee1b3eda001ab063f3 100644 (file)
--- a/src/client_side.cc
+++ b/src/client_side.cc
@@ -253,7 +253,7 @@ ConnStateData::readSomeData()
 
     typedef CommCbMemFunT<ConnStateData, CommIoCbParams> Dialer;
     reader = JobCallback(33, 5, Dialer, this, ConnStateData::clientReadRequest);
-    comm_read(clientConnection, in.buf.rawSpace(in.buf.spaceSize()), in.buf.spaceSize()-1, reader);
+    comm_read(clientConnection, in.buf, reader);
 }
 
 void
@@ -1563,7 +1563,7 @@ ConnStateData::readNextRequest()
 
     fd_note(clientConnection->fd, "Idle client: Waiting for next request");
     /**
-     * Set the timeout BEFORE calling clientReadRequest().
+     * Set the timeout BEFORE calling readSomeData().
      */
     typedef CommCbMemFunT<ConnStateData, CommTimeoutCbParams> TimeoutDialer;
     AsyncCall::Pointer timeoutCall = JobCallback(33, 5,
@@ -3006,7 +3006,7 @@ ConnStateData::clientReadRequest(const CommIoCbParams &io)
             kb_incr(&(statCounter.client_http.kbytes_in), io.size);
 
             // may comm_close or setReplyToError
-            if (!handleReadData(io.buf, io.size))
+            if (!handleReadData(io.buf2))
                 return;
 
         } else if (io.size == 0) {
@@ -3067,12 +3067,9 @@ ConnStateData::clientReadRequest(const CommIoCbParams &io)
  * \retval true  we did not call comm_close or setReplyToError
  */
 bool
-ConnStateData::handleReadData(char *buf, size_t size)
+ConnStateData::handleReadData(SBuf *buf)
 {
-    // XXX: make this a no-op when buf given is the MemBlob free space.
-    assert(buf == in.buf.rawSpace(1));
-    assert(size <= in.buf.spaceSize());
-    in.buf.append(buf, size);
+    assert(buf == &in.buf); // XXX: make this abort the transaction if this fails
 
     // if we are reading a body, stuff data into the body pipe
     if (bodyPipe != NULL)
@@ -3623,14 +3620,13 @@ httpsSslBumpAccessCheckDone(allow_t answer, void *data)
 
         // fake a CONNECT request to force connState to tunnel
         static char ip[MAX_IPSTRLEN];
-        static char reqStr[MAX_IPSTRLEN + 80];
         connState->clientConnection->local.toUrl(ip, sizeof(ip));
-        snprintf(reqStr, sizeof(reqStr), "CONNECT %s HTTP/1.1\r\nHost: %s\r\n\r\n", ip, ip);
-        bool ret = connState->handleReadData(reqStr, strlen(reqStr));
+        SBuf reqStr;
+        reqStr.append("CONNECT ").append(ip).append(" HTTP/1.1\r\nHost: ").append(ip).append("\r\n\r\n");
+        bool ret = connState->handleReadData(&reqStr);
         if (ret)
             ret = connState->clientParseRequests();
-
-        if (!ret) {
+        else {
             debugs(33, 2, HERE << "Failed to start fake CONNECT request for ssl bumped connection: " << connState->clientConnection);
             connState->clientConnection->close();
         }

diff --git a/src/client_side.h b/src/client_side.h
index ddd6901b90dc1a946ffee93b48f5f3d18c5b6521..70f56b89b3d11ae59ad21b40b36774ec046242ad 100644 (file)
--- a/src/client_side.h
+++ b/src/client_side.h
@@ -289,7 +289,7 @@ public:
     virtual void noteMoreBodySpaceAvailable(BodyPipe::Pointer);
     virtual void noteBodyConsumerAborted(BodyPipe::Pointer);
 
-    bool handleReadData(char *buf, size_t size);
+    bool handleReadData(SBuf *buf);
     bool handleRequestBodyData();
 
     /**

diff --git a/src/comm.cc b/src/comm.cc
index 48a4dfc64cf4b62fdfaa4cedcc278647bc350d01..73e4595487e4f231d0a913f87a4232949866ca0e 100644 (file)
--- a/src/comm.cc
+++ b/src/comm.cc
@@ -130,8 +130,19 @@ commHandleRead(int fd, void *data)
     ++ statCounter.syscalls.sock.reads;
     errno = 0;
     int retval;
-    retval = FD_READ_METHOD(fd, ccb->buf, ccb->size);
-    debugs(5, 3, "comm_read_try: FD " << fd << ", size " << ccb->size << ", retval " << retval << ", errno " << errno);
+    if (ccb->buf) {
+        retval = FD_READ_METHOD(fd, ccb->buf, ccb->size);
+        debugs(5, 3, "char FD " << fd << ", size " << ccb->size << ", retval " << retval << ", errno " << errno);
+    } else {
+        assert(ccb->buf2 != NULL);
+        SBuf::size_type sz = ccb->buf2->spaceSize();
+        char *buf = ccb->buf2->rawSpace(sz);
+        retval = FD_READ_METHOD(fd, buf, sz-1); // blocking synchronous read(2)
+        if (retval > 0) {
+            ccb->buf2->append(buf, retval);
+        }
+        debugs(5, 3, "SBuf FD " << fd << ", size " << sz << ", retval " << retval << ", errno " << errno);
+    }
 
     if (retval < 0 && !ignoreErrno(errno)) {
         debugs(5, 3, "comm_read_try: scheduling COMM_ERROR");
@@ -182,6 +193,36 @@ comm_read(const Comm::ConnectionPointer &conn, char *buf, int size, AsyncCall::P
     Comm::SetSelect(conn->fd, COMM_SELECT_READ, commHandleRead, ccb, 0);
 }
 
+/**
+ * Queue a read. handler/handler_data are called when the read
+ * completes, on error, or on file descriptor close.
+ */
+void
+comm_read(const Comm::ConnectionPointer &conn, SBuf &buf, AsyncCall::Pointer &callback)
+{
+    debugs(5, 5, "comm_read, queueing read for " << conn << "; asynCall " << callback);
+
+    /* Make sure we are open and not closing */
+    assert(Comm::IsConnOpen(conn));
+    assert(!fd_table[conn->fd].closing());
+    Comm::IoCallback *ccb = COMMIO_FD_READCB(conn->fd);
+
+    // Make sure we are either not reading or just passively monitoring.
+    // Active/passive conflicts are OK and simply cancel passive monitoring.
+    if (ccb->active()) {
+        // if the assertion below fails, we have an active comm_read conflict
+        assert(fd_table[conn->fd].halfClosedReader != NULL);
+        commStopHalfClosedMonitor(conn->fd);
+        assert(!ccb->active());
+    }
+    ccb->conn = conn;
+    ccb->buf2 = &buf;
+
+    /* Queue the read */
+    ccb->setCallback(Comm::IOCB_READ, callback, NULL, NULL, buf.spaceSize());
+    Comm::SetSelect(conn->fd, COMM_SELECT_READ, commHandleRead, ccb, 0);
+}
+
 /**
  * Empty the read buffers
  *

diff --git a/src/comm.h b/src/comm.h
index 8c5b04ea91af5a30721dccb39d55cacaddf4c2f5..feb8c367886eaa7af147ca2b9df515a4dafb4892 100644 (file)
--- a/src/comm.h
+++ b/src/comm.h
@@ -79,8 +79,8 @@ void comm_remove_close_handler(int fd, AsyncCall::Pointer &);
 
 int comm_has_pending_read_callback(int fd);
 bool comm_monitors_read(int fd);
-//void comm_read(const Comm::ConnectionPointer &conn, char *buf, int len, IOCB *handler, void *data);
 void comm_read(const Comm::ConnectionPointer &conn, char *buf, int len, AsyncCall::Pointer &callback);
+void comm_read(const Comm::ConnectionPointer &conn, SBuf &buf, AsyncCall::Pointer &callback);
 void comm_read_cancel(int fd, IOCB *callback, void *data);
 void comm_read_cancel(int fd, AsyncCall::Pointer &callback);
 int comm_udp_recvfrom(int fd, void *buf, size_t len, int flags, Ip::Address &from);

diff --git a/src/comm/IoCallback.cc b/src/comm/IoCallback.cc
index 67b8d039e6a4478c0021a58aacea402d6bbeced6..39f359429aadd3d11f865587379cf3542936ff4e 100644 (file)
--- a/src/comm/IoCallback.cc
+++ b/src/comm/IoCallback.cc
@@ -89,6 +89,7 @@ void
 Comm::IoCallback::reset()
 {
     conn = NULL;
+    buf2 = NULL; // we do not own this buffer.
     if (freefunc) {
         freefunc(buf);
         buf = NULL;
@@ -109,7 +110,7 @@ Comm::IoCallback::finish(comm_err_t code, int xerrn)
     assert(active());
 
     /* free data */
-    if (freefunc) {
+    if (freefunc && buf) {
         freefunc(buf);
         buf = NULL;
         freefunc = NULL;
@@ -120,6 +121,7 @@ Comm::IoCallback::finish(comm_err_t code, int xerrn)
         Params &params = GetCommParams<Params>(callback);
         if (conn != NULL) params.fd = conn->fd; // for legacy write handlers...
         params.conn = conn;
+        params.buf2 = buf2;
         params.buf = buf;
         params.size = offset;
         params.flag = code;

diff --git a/src/comm/IoCallback.h b/src/comm/IoCallback.h
index 85f021e371de614b14482a6052b478dc200f2c82..5d50c095786205b47a09e44447bc6f537b0ecd58 100644 (file)
--- a/src/comm/IoCallback.h
+++ b/src/comm/IoCallback.h
@@ -4,6 +4,7 @@
 #include "base/AsyncCall.h"
 #include "comm/forward.h"
 #include "comm_err_t.h"
+#include "SBuf.h"
 #include "typedefs.h"
 
 namespace Comm
@@ -23,6 +24,14 @@ public:
     iocb_type type;
     Comm::ConnectionPointer conn;
     AsyncCall::Pointer callback;
+
+    /// Buffer to store read(2) into when set.
+    // This is a pointer to the Jobs buffer rather than an SBuf using
+    // the same store since we cannot know when or how the Job will
+    // alter its SBuf while we are reading.
+    SBuf *buf2;
+
+    // Legacy c-string buffers used when buf2 is unset.
     char *buf;
     FREE *freefunc;
     int size;

diff --git a/src/tests/stub_client_side.cc b/src/tests/stub_client_side.cc
index f2173d0ced7551c648e8e0631f751a9928a527b6..d1f9ab1a3a14ff57a351f4a95305332c187e0eb0 100644 (file)
--- a/src/tests/stub_client_side.cc
+++ b/src/tests/stub_client_side.cc
@@ -51,7 +51,7 @@ void ConnStateData::stopSending(const char *error) STUB
 void ConnStateData::expectNoForwarding() STUB
 void ConnStateData::noteMoreBodySpaceAvailable(BodyPipe::Pointer) STUB
 void ConnStateData::noteBodyConsumerAborted(BodyPipe::Pointer) STUB
-bool ConnStateData::handleReadData(char *buf, size_t size) STUB_RETVAL(false)
+bool ConnStateData::handleReadData(SBuf *buf) STUB_RETVAL(false)
 bool ConnStateData::handleRequestBodyData() STUB_RETVAL(false)
 void ConnStateData::pinConnection(const Comm::ConnectionPointer &pinServerConn, HttpRequest *request, CachePeer *peer, bool auth) STUB
 void ConnStateData::unpinConnection() STUB