Bug 2907: high CPU usage on CONNECT when using delay pools

When delay pools are active on a CONNECT tunnel and teh pool is drained
the I/O loop cycles very often transferring 1 byte until the pool is
topped-up at the end of the second.

Instead of looping constantly trying to read 1 byte at a time, add an
asynchronous event to wait for a few I/O cycles or until more bytes can
be read.

To protect against infinite loops of waiting when many tunnels are
competing for the pool allowance we only delay for a limited number of
loops before allowing at least 1 byte through. Also, the amount of time
waited is an odd fraction of 1 second so re-tries naturally spread
across any given second fairly, with connections rotating closer or
further from the time when pool topup happens.

diff --git a/src/tunnel.cc b/src/tunnel.cc
index fec72681adc3a8c78eb69a678570b5d7b5c7b120..0c5b297c1fe28155520c1ab0ca71d621886c6564 100644 (file)
--- a/src/tunnel.cc
+++ b/src/tunnel.cc
@@ -12,6 +12,7 @@
 #include "acl/FilledChecklist.h"
 #include "base/CbcPointer.h"
 #include "CachePeer.h"
+#include "cbdata.h"
 #include "client_side.h"
 #include "client_side_request.h"
 #include "comm.h"
@@ -115,7 +116,7 @@ public:
     {
 
     public:
-        Connection() : len (0), buf ((char *)xmalloc(SQUID_TCP_SO_RCVBUF)), size_ptr(NULL) {}
+        Connection() : len (0), buf ((char *)xmalloc(SQUID_TCP_SO_RCVBUF)), size_ptr(NULL), delayedLoops(0) {}
 
         ~Connection();
 
@@ -137,7 +138,7 @@ public:
         int64_t *size_ptr;      /* pointer to size in an ConnStateData for logging */
 
         Comm::ConnectionPointer conn;    ///< The currently connected connection.
-
+        uint8_t delayedLoops; ///< how many times a read on this connection has been postponed.
     private:
 #if USE_DELAY_POOLS
 
@@ -331,6 +332,7 @@ void
 TunnelStateData::readServer(char *, size_t len, Comm::Flag errcode, int xerrno)
 {
     debugs(26, 3, HERE << server.conn << ", read " << len << " bytes, err=" << errcode);
+    server.delayedLoops=0;
 
     /*
      * Bail out early on Comm::ERR_CLOSING
@@ -476,6 +478,7 @@ void
 TunnelStateData::readClient(char *, size_t len, Comm::Flag errcode, int xerrno)
 {
     debugs(26, 3, HERE << client.conn << ", read " << len << " bytes, err=" << errcode);
+    client.delayedLoops=0;
 
     /*
      * Bail out early on Comm::ERR_CLOSING
@@ -676,13 +679,53 @@ TunnelStateData::Connection::closeIfOpen()
         conn->close();
 }
 
+static void
+tunnelDelayedClientRead(void *data)
+{
+    if (!data)
+        return;
+    TunnelStateData *tunnel = NULL;
+    static_cast<generic_cbdata*>(data)->unwrap(&tunnel);
+    if (!tunnel)
+        return;
+    static uint64_t counter=0;
+    debugs(26, 0, "Client read(2) delayed " << ++counter << " times");
+    tunnel->copyRead(tunnel->client, TunnelStateData::ReadClient);
+}
+
+static void
+tunnelDelayedServerRead(void *data)
+{
+    if (!data)
+        return;
+    TunnelStateData *tunnel = NULL;
+    static_cast<generic_cbdata*>(data)->unwrap(&tunnel);
+    if (!tunnel)
+        return;
+    static uint64_t counter=0;
+    debugs(26, 0, "Server read(2) delayed " << ++counter << " times");
+    tunnel->copyRead(tunnel->server, TunnelStateData::ReadServer);
+}
+
 void
 TunnelStateData::copyRead(Connection &from, IOCB *completion)
 {
     assert(from.len == 0);
+    // If only the minimum permitted read size is going to be attempted
+    // then we schedule an event to try again in a few I/O cycles.
+    // Allow at least 1 byte to be read every (0.3*10) seconds.
+    int bw = from.bytesWanted(1, SQUID_TCP_SO_RCVBUF);
+    if (bw == 1 && ++from.delayedLoops < 10) {
+        if (completion == TunnelStateData::ReadServer)
+            eventAdd("tunnelDelayedServerRead", &tunnelDelayedServerRead, new generic_cbdata(this), 0.3, true);
+        else
+            eventAdd("tunnelDelayedClientRead", &tunnelDelayedClientRead, new generic_cbdata(this), 0.3, true);
+        return;
+    }
+
     AsyncCall::Pointer call = commCbCall(5,4, "TunnelBlindCopyReadHandler",
                                          CommIoCbPtrFun(completion, this));
-    comm_read(from.conn, from.buf, from.bytesWanted(1, SQUID_TCP_SO_RCVBUF), call);
+    comm_read(from.conn, from.buf, bw, call);
 }
 
 void