Add a receive-buffer-size directive

... which can be used to set the size of the previously statically
allocated buffer in handle_established_connection().

Increasing the buffer size increases Privoxy's memory usage but
can lower the number of context switches and thereby reduce the
cpu usage and potentially increase the throughput.

This is mostly relevant for fast network connections and
large downloads that don't require filtering.

Currently BUFFER_SIZE is kept as default and lower limit
but the default should be increased after some more testing.

A dtrace command like:
sudo dtrace -n 'syscall::read:return /execname == "privoxy"/ { @[execname] = llquantize(arg0, 10, 0, 5, 20); @m = max(arg0)}'
can be used to properly tune the receive-buffer-size.

If the buffer is too large it will increase Privoxy's memory
footprint without any benefit. As the memory is (currently)
cleared before using it, a buffer that is too large can
actually reduce the throughput.

Things could be improved further by upwards scaling the buffer
dynamically based on how much of the previous allocation
was actually used.

Additionally the buffer should be referenced through csp and
also be used for other receive-related functions.

Measured throughput when using four connections to
constantly request a 10 MB file:

~320 MB/s with the default
~400 MB/s with "receive-buffer-size    8192"
~490 MB/s with "receive-buffer-size   16384"
~610 MB/s with "receive-buffer-size   32768"
~700 MB/s with "receive-buffer-size   65536"
~755 MB/s with "receive-buffer-size  131072"
~795 MB/s with "receive-buffer-size  262144"
~804 MB/s with "receive-buffer-size  524288"
~798 MB/s with "receive-buffer-size 1048576"
~780 MB/s with "receive-buffer-size 2097152"

Sponsored by: Robert Klemme

diff --git a/jcc.c b/jcc.c
index b335ca9..dd8b765 100644 (file)
--- a/jcc.c
+++ b/jcc.c
@@ -1,4 +1,4 @@
-const char jcc_rcs[] = "$Id: jcc.c,v 1.456 2017/05/25 11:16:56 fabiankeil Exp $";
+const char jcc_rcs[] = "$Id: jcc.c,v 1.457 2017/05/25 11:17:21 fabiankeil Exp $";

 /*********************************************************************
  *
  * File        :  $Source: /cvsroot/ijbswa/current/jcc.c,v $
 /*********************************************************************
  *
  * File        :  $Source: /cvsroot/ijbswa/current/jcc.c,v $


@@ -1962,7 +1962,7 @@ static int send_http_request(struct client_state *csp)
 static void handle_established_connection(struct client_state *csp,
                                           const struct forward_spec *fwd)
 {
 static void handle_established_connection(struct client_state *csp,
                                           const struct forward_spec *fwd)
 {

-   char buf[BUFFER_SIZE];
+   char *receive_buffer;

    char *hdr;
    char *p;
    int n;
    char *hdr;
    char *p;
    int n;


@@ -1985,8 +1985,17 @@ static void handle_established_connection(struct client_state *csp,
 #ifdef FEATURE_CONNECTION_KEEP_ALIVE
    int watch_client_socket;
 #endif
 #ifdef FEATURE_CONNECTION_KEEP_ALIVE
    int watch_client_socket;
 #endif

+   const size_t receive_buffer_size = csp->config->receive_buffer_size;

 
 

-   memset(buf, 0, sizeof(buf));
+   receive_buffer = zalloc(receive_buffer_size + 1);
+   if (receive_buffer == NULL)
+   {
+      log_error(LOG_LEVEL_ERROR,
+         "Out of memory. Failed to allocate the receive buffer.");
+      rsp = cgi_error_memory();
+      send_crunch_response(csp, rsp);
+      return;
+   }

 
    http = csp->http;
 
 
    http = csp->http;
 


@@ -2110,6 +2119,7 @@ static void handle_established_connection(struct client_state *csp,
             send_crunch_response(csp, error_response(csp, "connection-timeout"));
          }
          mark_server_socket_tainted(csp);
             send_crunch_response(csp, error_response(csp, "connection-timeout"));
          }
          mark_server_socket_tainted(csp);

+         freez(receive_buffer);

          return;
       }
       else if (n < 0)
          return;
       }
       else if (n < 0)


@@ -2120,6 +2130,7 @@ static void handle_established_connection(struct client_state *csp,
          log_error(LOG_LEVEL_ERROR, "select() failed!: %E");
 #endif
          mark_server_socket_tainted(csp);
          log_error(LOG_LEVEL_ERROR, "select() failed!: %E");
 #endif
          mark_server_socket_tainted(csp);

+         freez(receive_buffer);

          return;
       }
 
          return;
       }
 


@@ -2146,7 +2157,7 @@ static void handle_established_connection(struct client_state *csp,
       if (FD_ISSET(csp->cfd, &rfds))
 #endif /* def HAVE_POLL*/
       {
       if (FD_ISSET(csp->cfd, &rfds))
 #endif /* def HAVE_POLL*/
       {

-         int max_bytes_to_read = sizeof(buf) - 1;
+         int max_bytes_to_read = (int)receive_buffer_size;

 
 #ifdef FEATURE_CONNECTION_KEEP_ALIVE
          if ((csp->flags & CSP_FLAG_CLIENT_REQUEST_COMPLETELY_READ))
 
 #ifdef FEATURE_CONNECTION_KEEP_ALIVE
          if ((csp->flags & CSP_FLAG_CLIENT_REQUEST_COMPLETELY_READ))


@@ -2180,7 +2191,7 @@ static void handle_established_connection(struct client_state *csp,
          }
          if (csp->expected_client_content_length != 0)
          {
          }
          if (csp->expected_client_content_length != 0)
          {

-            if (csp->expected_client_content_length < (sizeof(buf) - 1))
+            if (csp->expected_client_content_length < receive_buffer_size)

             {
                max_bytes_to_read = (int)csp->expected_client_content_length;
             }
             {
                max_bytes_to_read = (int)csp->expected_client_content_length;
             }


@@ -2188,10 +2199,10 @@ static void handle_established_connection(struct client_state *csp,
                "Waiting for up to %d bytes from the client.",
                max_bytes_to_read);
          }
                "Waiting for up to %d bytes from the client.",
                max_bytes_to_read);
          }

-         assert(max_bytes_to_read < sizeof(buf));
+         assert(max_bytes_to_read <= receive_buffer_size);

 #endif /* def FEATURE_CONNECTION_KEEP_ALIVE */
 
 #endif /* def FEATURE_CONNECTION_KEEP_ALIVE */
 

-         len = read_socket(csp->cfd, buf, max_bytes_to_read);
+         len = read_socket(csp->cfd, receive_buffer, max_bytes_to_read);

 
          if (len <= 0)
          {
 
          if (len <= 0)
          {


@@ -2218,10 +2229,11 @@ static void handle_established_connection(struct client_state *csp,
          }
 #endif /* def FEATURE_CONNECTION_KEEP_ALIVE */
 
          }
 #endif /* def FEATURE_CONNECTION_KEEP_ALIVE */
 

-         if (write_socket(csp->server_connection.sfd, buf, (size_t)len))
+         if (write_socket(csp->server_connection.sfd, receive_buffer, (size_t)len))

          {
             log_error(LOG_LEVEL_ERROR, "write to: %s failed: %E", http->host);
             mark_server_socket_tainted(csp);
          {
             log_error(LOG_LEVEL_ERROR, "write to: %s failed: %E", http->host);
             mark_server_socket_tainted(csp);

+            freez(receive_buffer);

             return;
          }
          continue;
             return;
          }
          continue;


@@ -2255,12 +2267,13 @@ static void handle_established_connection(struct client_state *csp,
             log_error(LOG_LEVEL_CONNECT,
                "The server still wants to talk, but the client hung up on us.");
             mark_server_socket_tainted(csp);
             log_error(LOG_LEVEL_CONNECT,
                "The server still wants to talk, but the client hung up on us.");
             mark_server_socket_tainted(csp);

+            freez(receive_buffer);

             return;
 #endif /* def _WIN32 */
          }
 #endif /* def FEATURE_CONNECTION_KEEP_ALIVE */
 
             return;
 #endif /* def _WIN32 */
          }
 #endif /* def FEATURE_CONNECTION_KEEP_ALIVE */
 

-         len = read_socket(csp->server_connection.sfd, buf, sizeof(buf) - 1);
+         len = read_socket(csp->server_connection.sfd, receive_buffer, (int)receive_buffer_size);

 
          if (len < 0)
          {
 
          if (len < 0)
          {


@@ -2275,6 +2288,7 @@ static void handle_established_connection(struct client_state *csp,
                 */
                log_error(LOG_LEVEL_ERROR,
                   "CONNECT already confirmed. Unable to tell the client about the problem.");
                 */
                log_error(LOG_LEVEL_ERROR,
                   "CONNECT already confirmed. Unable to tell the client about the problem.");

+               freez(receive_buffer);

                return;
             }
             else if (byte_count)
                return;
             }
             else if (byte_count)


@@ -2289,6 +2303,7 @@ static void handle_established_connection(struct client_state *csp,
                log_error(LOG_LEVEL_ERROR, "Already forwarded the original headers. "
                   "Unable to tell the client about the problem.");
                mark_server_socket_tainted(csp);
                log_error(LOG_LEVEL_ERROR, "Already forwarded the original headers. "
                   "Unable to tell the client about the problem.");
                mark_server_socket_tainted(csp);

+               freez(receive_buffer);

                return;
             }
             /*
                return;
             }
             /*


@@ -2301,7 +2316,7 @@ static void handle_established_connection(struct client_state *csp,
 #ifdef FEATURE_CONNECTION_KEEP_ALIVE
          if (csp->flags & CSP_FLAG_CHUNKED)
          {
 #ifdef FEATURE_CONNECTION_KEEP_ALIVE
          if (csp->flags & CSP_FLAG_CHUNKED)
          {

-            if ((len >= 5) && !memcmp(buf+len-5, "0\r\n\r\n", 5))
+            if ((len >= 5) && !memcmp(receive_buffer+len-5, "0\r\n\r\n", 5))

             {
                /* XXX: this is a temporary hack */
                log_error(LOG_LEVEL_CONNECT,
             {
                /* XXX: this is a temporary hack */
                log_error(LOG_LEVEL_CONNECT,


@@ -2314,11 +2329,23 @@ static void handle_established_connection(struct client_state *csp,
          reading_done:
 #endif  /* FEATURE_CONNECTION_KEEP_ALIVE */
 
          reading_done:
 #endif  /* FEATURE_CONNECTION_KEEP_ALIVE */
 

+         /*
+          * This is guaranteed by allocating with zalloc_or_die()
+          * and never (intentionally) writing to the last byte.
+          *
+          * receive_buffer_size is the size of the part of the
+          * buffer we intentionally write to, but we actually
+          * allocated receive_buffer_size+1 bytes so the assertion
+          * stays within the allocated range.
+          */
+         assert(receive_buffer[receive_buffer_size] == '\0');
+

          /*
           * Add a trailing zero to let be able to use string operations.
           * XXX: do we still need this with filter_popups gone?
           */
          /*
           * Add a trailing zero to let be able to use string operations.
           * XXX: do we still need this with filter_popups gone?
           */

-         buf[len] = '\0';
+         assert(len <= receive_buffer_size);
+         receive_buffer[len] = '\0';

 
          /*
           * Normally, this would indicate that we've read
 
          /*
           * Normally, this would indicate that we've read


@@ -2397,6 +2424,7 @@ static void handle_established_connection(struct client_state *csp,
                      freez(hdr);
                      freez(p);
                      mark_server_socket_tainted(csp);
                      freez(hdr);
                      freez(p);
                      mark_server_socket_tainted(csp);

+                     freez(receive_buffer);

                      return;
                   }
 
                      return;
                   }
 


@@ -2411,8 +2439,8 @@ static void handle_established_connection(struct client_state *csp,
              * This is NOT the body, so
              * Let's pretend the server just sent us a blank line.
              */
              * This is NOT the body, so
              * Let's pretend the server just sent us a blank line.
              */

-            snprintf(buf, sizeof(buf), "\r\n");
-            len = (int)strlen(buf);
+            snprintf(receive_buffer, receive_buffer_size, "\r\n");
+            len = (int)strlen(receive_buffer);

 
             /*
              * Now, let the normal header parsing algorithm below do its
 
             /*
              * Now, let the normal header parsing algorithm below do its


@@ -2436,7 +2464,7 @@ static void handle_established_connection(struct client_state *csp,
                 * has been reached, switch to non-filtering mode, i.e. make & write the
                 * header, flush the iob and buf, and get out of the way.
                 */
                 * has been reached, switch to non-filtering mode, i.e. make & write the
                 * header, flush the iob and buf, and get out of the way.
                 */

-               if (add_to_iob(csp->iob, csp->config->buffer_limit, buf, len))
+               if (add_to_iob(csp->iob, csp->config->buffer_limit, receive_buffer, len))

                {
                   size_t hdrlen;
                   long flushed;
                {
                   size_t hdrlen;
                   long flushed;


@@ -2455,18 +2483,20 @@ static void handle_established_connection(struct client_state *csp,
                      rsp = cgi_error_memory();
                      send_crunch_response(csp, rsp);
                      mark_server_socket_tainted(csp);
                      rsp = cgi_error_memory();
                      send_crunch_response(csp, rsp);
                      mark_server_socket_tainted(csp);

+                     freez(receive_buffer);

                      return;
                   }
                   hdrlen = strlen(hdr);
 
                   if (write_socket(csp->cfd, hdr, hdrlen)
                    || ((flushed = flush_socket(csp->cfd, csp->iob)) < 0)
                      return;
                   }
                   hdrlen = strlen(hdr);
 
                   if (write_socket(csp->cfd, hdr, hdrlen)
                    || ((flushed = flush_socket(csp->cfd, csp->iob)) < 0)

-                   || (write_socket(csp->cfd, buf, (size_t)len)))
+                   || (write_socket(csp->cfd, receive_buffer, (size_t)len)))

                   {
                      log_error(LOG_LEVEL_CONNECT,
                         "Flush header and buffers to client failed: %E");
                      freez(hdr);
                      mark_server_socket_tainted(csp);
                   {
                      log_error(LOG_LEVEL_CONNECT,
                         "Flush header and buffers to client failed: %E");
                      freez(hdr);
                      mark_server_socket_tainted(csp);

+                     freez(receive_buffer);

                      return;
                   }
 
                      return;
                   }
 


@@ -2483,10 +2513,11 @@ static void handle_established_connection(struct client_state *csp,
             }
             else
             {
             }
             else
             {

-               if (write_socket(csp->cfd, buf, (size_t)len))
+               if (write_socket(csp->cfd, receive_buffer, (size_t)len))

                {
                   log_error(LOG_LEVEL_ERROR, "write to client failed: %E");
                   mark_server_socket_tainted(csp);
                {
                   log_error(LOG_LEVEL_ERROR, "write to client failed: %E");
                   mark_server_socket_tainted(csp);

+                  freez(receive_buffer);

                   return;
                }
             }
                   return;
                }
             }


@@ -2500,12 +2531,13 @@ static void handle_established_connection(struct client_state *csp,
              * Buffer up the data we just read.  If that fails, there's
              * little we can do but send our static out-of-memory page.
              */
              * Buffer up the data we just read.  If that fails, there's
              * little we can do but send our static out-of-memory page.
              */

-            if (add_to_iob(csp->iob, csp->config->buffer_limit, buf, len))
+            if (add_to_iob(csp->iob, csp->config->buffer_limit, receive_buffer, len))

             {
                log_error(LOG_LEVEL_ERROR, "Out of memory while looking for end of server headers.");
                rsp = cgi_error_memory();
                send_crunch_response(csp, rsp);
                mark_server_socket_tainted(csp);
             {
                log_error(LOG_LEVEL_ERROR, "Out of memory while looking for end of server headers.");
                rsp = cgi_error_memory();
                send_crunch_response(csp, rsp);
                mark_server_socket_tainted(csp);

+               freez(receive_buffer);

                return;
             }
 
                return;
             }
 


@@ -2526,6 +2558,7 @@ static void handle_established_connection(struct client_state *csp,
                   write_socket(csp->cfd, INVALID_SERVER_HEADERS_RESPONSE,
                      strlen(INVALID_SERVER_HEADERS_RESPONSE));
                   mark_server_socket_tainted(csp);
                   write_socket(csp->cfd, INVALID_SERVER_HEADERS_RESPONSE,
                      strlen(INVALID_SERVER_HEADERS_RESPONSE));
                   mark_server_socket_tainted(csp);

+                  freez(receive_buffer);

                   return;
                }
                else
                   return;
                }
                else


@@ -2571,6 +2604,7 @@ static void handle_established_connection(struct client_state *csp,
                }
                free_http_request(http);
                mark_server_socket_tainted(csp);
                }
                free_http_request(http);
                mark_server_socket_tainted(csp);

+               freez(receive_buffer);

                return;
             }
 
                return;
             }
 


@@ -2611,6 +2645,7 @@ static void handle_established_connection(struct client_state *csp,
                   strlen(INVALID_SERVER_HEADERS_RESPONSE));
                free_http_request(http);
                mark_server_socket_tainted(csp);
                   strlen(INVALID_SERVER_HEADERS_RESPONSE));
                free_http_request(http);
                mark_server_socket_tainted(csp);

+               freez(receive_buffer);

                return;
             }
             hdr = list_to_text(csp->headers);
                return;
             }
             hdr = list_to_text(csp->headers);


@@ -2645,6 +2680,7 @@ static void handle_established_connection(struct client_state *csp,
                 */
                 freez(hdr);
                 mark_server_socket_tainted(csp);
                 */
                 freez(hdr);
                 mark_server_socket_tainted(csp);

+                freez(receive_buffer);

                 return;
             }
             /* Buffer and pcrs filter this if appropriate. */
                 return;
             }
             /* Buffer and pcrs filter this if appropriate. */


@@ -2675,6 +2711,7 @@ static void handle_established_connection(struct client_state *csp,
                    */
                   freez(hdr);
                   mark_server_socket_tainted(csp);
                    */
                   freez(hdr);
                   mark_server_socket_tainted(csp);

+                  freez(receive_buffer);

                   return;
                }
             }
                   return;
                }
             }


@@ -2699,14 +2736,17 @@ static void handle_established_connection(struct client_state *csp,
                write_socket(csp->cfd, INVALID_SERVER_HEADERS_RESPONSE,
                   strlen(INVALID_SERVER_HEADERS_RESPONSE));
                mark_server_socket_tainted(csp);
                write_socket(csp->cfd, INVALID_SERVER_HEADERS_RESPONSE,
                   strlen(INVALID_SERVER_HEADERS_RESPONSE));
                mark_server_socket_tainted(csp);

+               freez(receive_buffer);

                return;
             }
          }
          continue;
       }
       mark_server_socket_tainted(csp);
                return;
             }
          }
          continue;
       }
       mark_server_socket_tainted(csp);

+      freez(receive_buffer);

       return; /* huh? we should never get here */
    }
       return; /* huh? we should never get here */
    }

+   freez(receive_buffer);

 
    if (csp->content_length == 0)
    {
 
    if (csp->content_length == 0)
    {

diff --git a/loadcfg.c b/loadcfg.c
index 3382b6d..9ffd509 100644 (file)
--- a/loadcfg.c
+++ b/loadcfg.c
@@ -1,4 +1,4 @@
-const char loadcfg_rcs[] = "$Id: loadcfg.c,v 1.158 2017/05/25 11:16:56 fabiankeil Exp $";
+const char loadcfg_rcs[] = "$Id: loadcfg.c,v 1.159 2017/05/25 11:17:38 fabiankeil Exp $";

 /*********************************************************************
  *
  * File        :  $Source: /cvsroot/ijbswa/current/loadcfg.c,v $
 /*********************************************************************
  *
  * File        :  $Source: /cvsroot/ijbswa/current/loadcfg.c,v $


@@ -158,6 +158,7 @@ static struct file_list *current_configfile = NULL;
 #define hash_max_client_connections      3595884446U /* "max-client-connections" */
 #define hash_permit_access               3587953268U /* "permit-access" */
 #define hash_proxy_info_url              3903079059U /* "proxy-info-url" */
 #define hash_max_client_connections      3595884446U /* "max-client-connections" */
 #define hash_permit_access               3587953268U /* "permit-access" */
 #define hash_proxy_info_url              3903079059U /* "proxy-info-url" */

+#define hash_receive_buffer_size         2880297454U /* "receive-buffer-size */

 #define hash_single_threaded             4250084780U /* "single-threaded" */
 #define hash_socket_timeout              1809001761U /* "socket-timeout" */
 #define hash_split_large_cgi_forms        671658948U /* "split-large-cgi-forms" */
 #define hash_single_threaded             4250084780U /* "single-threaded" */
 #define hash_socket_timeout              1809001761U /* "socket-timeout" */
 #define hash_split_large_cgi_forms        671658948U /* "split-large-cgi-forms" */


@@ -597,6 +598,7 @@ struct configuration_spec * load_config(void)
     */
    config->multi_threaded            = 1;
    config->buffer_limit              = 4096 * 1024;
     */
    config->multi_threaded            = 1;
    config->buffer_limit              = 4096 * 1024;

+   config->receive_buffer_size       = BUFFER_SIZE;

    config->usermanual                = strdup_or_die(USER_MANUAL_URL);
    config->proxy_args                = strdup_or_die("");
    config->forwarded_connect_retries = 0;
    config->usermanual                = strdup_or_die(USER_MANUAL_URL);
    config->proxy_args                = strdup_or_die("");
    config->forwarded_connect_retries = 0;


@@ -1500,6 +1502,21 @@ struct configuration_spec * load_config(void)
             config->proxy_info_url = strdup_or_die(arg);
             break;
 
             config->proxy_info_url = strdup_or_die(arg);
             break;
 

+
+/* *************************************************************************
+ * receive-buffer-size n
+ * *************************************************************************/
+         case hash_receive_buffer_size :
+            config->receive_buffer_size = (size_t)parse_numeric_value(cmd, arg);
+            if (config->receive_buffer_size < BUFFER_SIZE)
+            {
+               log_error(LOG_LEVEL_INFO,
+                  "receive-buffer-size %d seems low and may cause problems."
+                  "Consider setting it to at least %d.",
+                  config->receive_buffer_size, BUFFER_SIZE);
+            }
+            break;
+

 /* *************************************************************************
  * single-threaded 0|1
  * *************************************************************************/
 /* *************************************************************************
  * single-threaded 0|1
  * *************************************************************************/

diff --git a/project.h b/project.h
index 1fc0bd4..65427e0 100644 (file)
--- a/project.h
+++ b/project.h
@@ -1,7 +1,7 @@
 #ifndef PROJECT_H_INCLUDED
 #define PROJECT_H_INCLUDED
 /** Version string. */
 #ifndef PROJECT_H_INCLUDED
 #define PROJECT_H_INCLUDED
 /** Version string. */

-#define PROJECT_H_VERSION "$Id: project.h,v 1.219 2017/01/23 16:10:28 fabiankeil Exp $"
+#define PROJECT_H_VERSION "$Id: project.h,v 1.220 2017/02/20 13:44:32 fabiankeil Exp $"

 /*********************************************************************
  *
  * File        :  $Source: /cvsroot/ijbswa/current/project.h,v $
 /*********************************************************************
  *
  * File        :  $Source: /cvsroot/ijbswa/current/project.h,v $


@@ -1348,6 +1348,9 @@ struct configuration_spec
    /** Size limit for IOB */
    size_t buffer_limit;
 
    /** Size limit for IOB */
    size_t buffer_limit;
 

+   /** Size of the receive buffer */
+   size_t receive_buffer_size;
+

 #ifdef FEATURE_TRUST
 
    /** The file name of the trust file. */
 #ifdef FEATURE_TRUST
 
    /** The file name of the trust file. */