Use smaller zlib objects when under memory pressure

We add a compression level argument to tor_zlib_new, and use it to
determine how much memory to allocate for the zlib object.  We use the
existing level by default, but shift to smaller levels for small
requests when we have been over 3/4 of our memory usage in the past
half-hour.

Closes ticket 11791.

changes/bug11791

@@ -0,0 +1,4 @@
+  o Minor features (directory, memory usage):
+    - When we have recently been under memory pressure (over 3/4 of
+      MaxMemInQueues is allocated), then allocate smaller zlib objects for
+      small requests. Closes ticket 11791.

src/common/torgzip.c

@@ -92,10 +92,27 @@ tor_zlib_get_header_version_str(void)
 
 /** Return the 'bits' value to tell zlib to use <b>method</b>.*/
 static INLINE int
-method_bits(compress_method_t method)
+method_bits(compress_method_t method, zlib_compression_level_t level)
 {
   /* Bits+16 means "use gzip" in zlib >= 1.2 */
-  return method == GZIP_METHOD ? 15+16 : 15;
+  const int flag = method == GZIP_METHOD ? 16 : 0;
+  switch (level) {
+    default:
+    case HIGH_COMPRESSION: return flag + 15;
+    case MEDIUM_COMPRESSION: return flag + 13;
+    case LOW_COMPRESSION: return flag + 11;
+  }
+}
+
+static INLINE int
+get_memlevel(zlib_compression_level_t level)
+{
+  switch (level) {
+    default:
+    case HIGH_COMPRESSION: return 8;
+    case MEDIUM_COMPRESSION: return 7;
+    case LOW_COMPRESSION: return 6;
+  }
 }
 
 /** @{ */
@@ -162,8 +179,9 @@ tor_gzip_compress(char **out, size_t *out_len,
   stream->avail_in = (unsigned int)in_len;
 
   if (deflateInit2(stream, Z_BEST_COMPRESSION, Z_DEFLATED,
-                   method_bits(method),
-                   8, Z_DEFAULT_STRATEGY) != Z_OK) {
+                   method_bits(method, HIGH_COMPRESSION),
+                   get_memlevel(HIGH_COMPRESSION),
+                   Z_DEFAULT_STRATEGY) != Z_OK) {
     log_warn(LD_GENERAL, "Error from deflateInit2: %s",
              stream->msg?stream->msg:"<no message>");
     goto err;
@@ -289,7 +307,7 @@ tor_gzip_uncompress(char **out, size_t *out_len,
   stream->avail_in = (unsigned int)in_len;
 
   if (inflateInit2(stream,
-                   method_bits(method)) != Z_OK) {
+                   method_bits(method, HIGH_COMPRESSION)) != Z_OK) {
     log_warn(LD_GENERAL, "Error from inflateInit2: %s",
              stream->msg?stream->msg:"<no message>");
     goto err;
@@ -315,7 +333,8 @@ tor_gzip_uncompress(char **out, size_t *out_len,
           log_warn(LD_BUG, "Error freeing gzip structures");
           goto err;
         }
-        if (inflateInit2(stream, method_bits(method)) != Z_OK) {
+        if (inflateInit2(stream,
+                         method_bits(method,HIGH_COMPRESSION)) != Z_OK) {
           log_warn(LD_GENERAL, "Error from second inflateInit2: %s",
                    stream->msg?stream->msg:"<no message>");
           goto err;
@@ -426,10 +445,11 @@ struct tor_zlib_state_t {
  * <b>compress</b>, it's for compression; otherwise it's for
  * decompression. */
 tor_zlib_state_t *
-tor_zlib_new(int compress, compress_method_t method)
+tor_zlib_new(int compress, compress_method_t method,
+             zlib_compression_level_t compression_level)
 {
   tor_zlib_state_t *out;
-  int bits;
+  int bits, memlevel;
 
   if (method == GZIP_METHOD && !is_gzip_supported()) {
     /* Old zlib version don't support gzip in inflateInit2 */
@@ -437,21 +457,29 @@ tor_zlib_new(int compress, compress_method_t method)
     return NULL;
  }
 
+ if (! compress) {
+   /* use this setting for decompression, since we might have the
+    * max number of window bits */
+   compression_level = HIGH_COMPRESSION;
+ }
+
  out = tor_malloc_zero(sizeof(tor_zlib_state_t));
  out->stream.zalloc = Z_NULL;
  out->stream.zfree = Z_NULL;
  out->stream.opaque = NULL;
  out->compress = compress;
- bits = method_bits(method);
+ bits = method_bits(method, compression_level);
+ memlevel = get_memlevel(compression_level);
  if (compress) {
    if (deflateInit2(&out->stream, Z_BEST_COMPRESSION, Z_DEFLATED,
-                    bits, 8, Z_DEFAULT_STRATEGY) != Z_OK)
+                    bits, memlevel,
+                    Z_DEFAULT_STRATEGY) != Z_OK)
      goto err;
  } else {
    if (inflateInit2(&out->stream, bits) != Z_OK)
      goto err;
  }
- out->allocation = tor_zlib_state_size_precalc(!compress, bits, 8);
+ out->allocation = tor_zlib_state_size_precalc(!compress, bits, memlevel);
 
  total_zlib_allocation += out->allocation;
 

src/common/torgzip.h

@@ -19,6 +19,15 @@ typedef enum {
   NO_METHOD=0, GZIP_METHOD=1, ZLIB_METHOD=2, UNKNOWN_METHOD=3
 } compress_method_t;
 
+/**
+ * Enumeration to define tradeoffs between memory usage and compression level.
+ * HIGH_COMPRESSION saves the most bandwidth; LOW_COMPRESSION saves the most
+ * memory.
+ **/
+typedef enum {
+  HIGH_COMPRESSION, MEDIUM_COMPRESSION, LOW_COMPRESSION
+} zlib_compression_level_t;
+
 int
 tor_gzip_compress(char **out, size_t *out_len,
                   const char *in, size_t in_len,
@@ -47,7 +56,8 @@ typedef enum {
 } tor_zlib_output_t;
 /** Internal state for an incremental zlib compression/decompression. */
 typedef struct tor_zlib_state_t tor_zlib_state_t;
-tor_zlib_state_t *tor_zlib_new(int compress, compress_method_t method);
+tor_zlib_state_t *tor_zlib_new(int compress, compress_method_t method,
+                               zlib_compression_level_t level);
 
 tor_zlib_output_t tor_zlib_process(tor_zlib_state_t *state,
                                    char **out, size_t *out_len,

src/or/config.c

@@ -2828,6 +2828,7 @@ options_validate(or_options_t *old_options, or_options_t *options,
   options->MaxMemInQueues =
     compute_real_max_mem_in_queues(options->MaxMemInQueues_raw,
                                    server_mode(options));
+  options->MaxMemInQueues_low_threshold = (options->MaxMemInQueues / 4) * 3;
 
   options->AllowInvalid_ = 0;
 

src/or/directory.c

@@ -20,6 +20,7 @@
 #include "networkstatus.h"
 #include "nodelist.h"
 #include "policies.h"
+#include "relay.h"
 #include "rendclient.h"
 #include "rendcommon.h"
 #include "rephist.h"
@@ -2521,6 +2522,24 @@ client_likes_consensus(networkstatus_t *v, const char *want_url)
   return (have >= need_at_least);
 }
 
+/** Return the compression level we should use for sending a compressed
+ * response of size <b>n_bytes</b>. */
+static zlib_compression_level_t
+choose_compression_level(ssize_t n_bytes)
+{
+  if (! have_been_under_memory_pressure()) {
+    return HIGH_COMPRESSION; /* we have plenty of RAM. */
+  } else if (n_bytes < 0) {
+    return HIGH_COMPRESSION; /* unknown; might be big. */
+  } else if (n_bytes < 1024) {
+    return LOW_COMPRESSION;
+  } else if (n_bytes < 2048) {
+    return MEDIUM_COMPRESSION;
+  } else {
+    return HIGH_COMPRESSION;
+  }
+}
+
 /** Helper function: called when a dirserver gets a complete HTTP GET
  * request.  Look for a request for a directory or for a rendezvous
  * service descriptor.  On finding one, write a response into
@@ -2703,7 +2722,7 @@ directory_handle_command_get(dir_connection_t *conn, const char *headers,
                                smartlist_len(dir_fps) == 1 ? lifetime : 0);
     conn->fingerprint_stack = dir_fps;
     if (! compressed)
-      conn->zlib_state = tor_zlib_new(0, ZLIB_METHOD);
+      conn->zlib_state = tor_zlib_new(0, ZLIB_METHOD, HIGH_COMPRESSION);
 
     /* Prime the connection with some data. */
     conn->dir_spool_src = DIR_SPOOL_NETWORKSTATUS;
@@ -2791,7 +2810,8 @@ directory_handle_command_get(dir_connection_t *conn, const char *headers,
 
     if (smartlist_len(items)) {
       if (compressed) {
-        conn->zlib_state = tor_zlib_new(1, ZLIB_METHOD);
+        conn->zlib_state = tor_zlib_new(1, ZLIB_METHOD,
+                                    choose_compression_level(estimated_len));
         SMARTLIST_FOREACH(items, const char *, c,
                  connection_write_to_buf_zlib(c, strlen(c), conn, 0));
         connection_write_to_buf_zlib("", 0, conn, 1);
@@ -2840,7 +2860,8 @@ directory_handle_command_get(dir_connection_t *conn, const char *headers,
     conn->fingerprint_stack = fps;
 
     if (compressed)
-      conn->zlib_state = tor_zlib_new(1, ZLIB_METHOD);
+      conn->zlib_state = tor_zlib_new(1, ZLIB_METHOD,
+                                      choose_compression_level(dlen));
 
     connection_dirserv_flushed_some(conn);
     goto done;
@@ -2908,7 +2929,8 @@ directory_handle_command_get(dir_connection_t *conn, const char *headers,
       }
       write_http_response_header(conn, -1, compressed, cache_lifetime);
       if (compressed)
-        conn->zlib_state = tor_zlib_new(1, ZLIB_METHOD);
+        conn->zlib_state = tor_zlib_new(1, ZLIB_METHOD,
+                                        choose_compression_level(dlen));
       /* Prime the connection with some data. */
       connection_dirserv_flushed_some(conn);
     }
@@ -2983,7 +3005,8 @@ directory_handle_command_get(dir_connection_t *conn, const char *headers,
 
     write_http_response_header(conn, compressed?-1:len, compressed, 60*60);
     if (compressed) {
-      conn->zlib_state = tor_zlib_new(1, ZLIB_METHOD);
+      conn->zlib_state = tor_zlib_new(1, ZLIB_METHOD,
+                                      choose_compression_level(len));
       SMARTLIST_FOREACH(certs, authority_cert_t *, c,
             connection_write_to_buf_zlib(c->cache_info.signed_descriptor_body,
                                          c->cache_info.signed_descriptor_len,

src/or/dirserv.c

@@ -3182,7 +3182,7 @@ connection_dirserv_add_networkstatus_bytes_to_outbuf(dir_connection_t *conn)
         if (uncompressing && ! conn->zlib_state &&
             conn->fingerprint_stack &&
             smartlist_len(conn->fingerprint_stack)) {
-          conn->zlib_state = tor_zlib_new(0, ZLIB_METHOD);
+          conn->zlib_state = tor_zlib_new(0, ZLIB_METHOD, HIGH_COMPRESSION);
         }
       }
       if (r) return r;

src/or/or.h

@@ -3514,6 +3514,8 @@ typedef struct {
   uint64_t MaxMemInQueues_raw;
   uint64_t MaxMemInQueues;/**< If we have more memory than this allocated
                             * for queues and buffers, run the OOM handler */
+  /** Above this value, consider ourselves low on RAM. */
+  uint64_t MaxMemInQueues_low_threshold;
 
   /** @name port booleans
    *

src/or/relay.c

@@ -2432,6 +2432,12 @@ cell_queues_get_total_allocation(void)
   return total_cells_allocated * packed_cell_mem_cost();
 }
 
+/** How long after we've been low on memory should we try to conserve it? */
+#define MEMORY_PRESSURE_INTERVAL (30*60)
+
+/** The time at which we were last low on memory. */
+static time_t last_time_under_memory_pressure = 0;
+
 /** Check whether we've got too much space used for cells.  If so,
  * call the OOM handler and return 1.  Otherwise, return 0. */
 STATIC int
@@ -2440,13 +2446,25 @@ cell_queues_check_size(void)
   size_t alloc = cell_queues_get_total_allocation();
   alloc += buf_get_total_allocation();
   alloc += tor_zlib_get_total_allocation();
-  if (alloc >= get_options()->MaxMemInQueues) {
-    circuits_handle_oom(alloc);
-    return 1;
+  if (alloc >= get_options()->MaxMemInQueues_low_threshold) {
+    last_time_under_memory_pressure = approx_time();
+    if (alloc >= get_options()->MaxMemInQueues) {
+      circuits_handle_oom(alloc);
+      return 1;
+    }
   }
   return 0;
 }
 
+/** Return true if we've been under memory pressure in the last
+ * MEMORY_PRESSURE_INTERVAL seconds. */
+int
+have_been_under_memory_pressure(void)
+{
+  return last_time_under_memory_pressure + MEMORY_PRESSURE_INTERVAL
+    < approx_time();
+}
+
 /**
  * Update the number of cells available on the circuit's n_chan or p_chan's
  * circuit mux.

src/or/relay.h

@@ -50,6 +50,8 @@ void clean_cell_pool(void);
 void dump_cell_pool_usage(int severity);
 size_t packed_cell_mem_cost(void);
 
+int have_been_under_memory_pressure(void);
+
 /* For channeltls.c */
 void packed_cell_free(packed_cell_t *cell);
 

src/test/test_buffers.c

@@ -611,7 +611,7 @@ test_buffers_zlib_impl(int finalize_with_nil)
   int done;
 
   buf = buf_new_with_capacity(128); /* will round up */
-  zlib_state = tor_zlib_new(1, ZLIB_METHOD);
+  zlib_state = tor_zlib_new(1, ZLIB_METHOD, HIGH_COMPRESSION);
 
   msg = tor_malloc(512);
   crypto_rand(msg, 512);
@@ -688,7 +688,7 @@ test_buffers_zlib_fin_at_chunk_end(void *arg)
   tt_uint_op(buf->head->datalen, OP_EQ, headerjunk);
   tt_uint_op(buf_datalen(buf), OP_EQ, headerjunk);
   /* Write an empty string, with finalization on. */
-  zlib_state = tor_zlib_new(1, ZLIB_METHOD);
+  zlib_state = tor_zlib_new(1, ZLIB_METHOD, HIGH_COMPRESSION);
   tt_int_op(write_to_buf_zlib(buf, zlib_state, "", 0, 1), OP_EQ, 0);
 
   in_len = buf_datalen(buf);

src/test/test_util.c

@@ -1820,7 +1820,7 @@ test_util_gzip(void *arg)
   tor_free(buf1);
   tor_free(buf2);
   tor_free(buf3);
-  state = tor_zlib_new(1, ZLIB_METHOD);
+  state = tor_zlib_new(1, ZLIB_METHOD, HIGH_COMPRESSION);
   tt_assert(state);
   cp1 = buf1 = tor_malloc(1024);
   len1 = 1024;