BUG/MEDIUM: compression: fix incompatibility between compression and large-buffers

The SLZ compression being stateless, it doesn't keep data between calls
and must be called with the largest blocks at once to be efficient. When
it was added, muxes were not there yet and chunks were parsed inside the
HTTP forwarding analyzer, which would call comp->add_data() for each chunk
of data. In case of small chunks, that would just not compress, so a local
buffer was added and data were temporarily stored there and effectively
compressed when calling ->flush().

But with the introduction of large buffers, it became a problem because
the locally allocated buffer can be too small if the data to be compressed
comes from a large one. One solution to this is to also pre-allocate a
large buffer, but since there needs to be one per thread and that some
users might use extremely large buffers on large systems, that could be
extremely expensive in terms of memory.

A better solution is to just simplify all of this as this complexity is
no longer needed at the era of muxes which extract chunks and make a
large HTX data block with them. That's what this patch does, and the
functions become as simple as their zlib counterparts, no longer require
a local per-thread buffer, and will be even faster thanks to the removal
of a long-unneeded extra copy.

This fix will need to be backported to 3.4 where large buffers were
introduced, maybe after a period of observation. An alternate fix in
the mean time could be to reject configs having both compression and
large buffers.

diff --git a/include/haproxy/compression-t.h b/include/haproxy/compression-t.h
index 39e4175710faaa36aad097576cbe73bba02d75b6..1b71cdc145d4e2aa4129a709ec7fc6cb2ebe4a7b 100644 (file)
--- a/include/haproxy/compression-t.h
+++ b/include/haproxy/compression-t.h
@@ -59,9 +59,6 @@ struct comp {
 struct comp_ctx {
 #if defined(USE_SLZ)
        struct slz_stream strm;
-       const void *direct_ptr; /* NULL or pointer to beginning of data */
-       int direct_len;         /* length of direct_ptr if not NULL */
-       struct buffer queued;   /* if not null, data already queued */
 #elif defined(USE_ZLIB)
        z_stream strm; /* zlib stream */
        void *zlib_deflate_state;

diff --git a/src/compression.c b/src/compression.c
index 308a1de3d9133d582536c9507fdfb411cc455d87..d0fceb15ddc087205b8e64b5d04527cf4decbae8 100644 (file)
--- a/src/compression.c
+++ b/src/compression.c
@@ -217,11 +217,7 @@ static inline int init_comp_ctx(struct comp_ctx **comp_ctx)
        *comp_ctx = pool_alloc(pool_comp_ctx);
        if (*comp_ctx == NULL)
                return -1;
-#if defined(USE_SLZ)
-       (*comp_ctx)->direct_ptr = NULL;
-       (*comp_ctx)->direct_len = 0;
-       (*comp_ctx)->queued = BUF_NULL;
-#elif defined(USE_ZLIB)
+#if defined(USE_ZLIB)
        _HA_ATOMIC_ADD(&zlib_used_memory, sizeof(struct comp_ctx));
        __ha_barrier_atomic_store();
 
@@ -367,42 +363,22 @@ static int comp_rfc1950_init(struct comp_ctx **comp_ctx, int level)
        return slz_rfc1950_init(&(*comp_ctx)->strm, !!level);
 }
 
-/* Return the size of consumed data or -1. The output buffer is unused at this
- * point, we only keep a reference to the input data or a copy of them if the
- * reference is already used.
+/* Return the size of consumed data. The caller is responsible for guaranteeing
+ * that <out> is large enough to keep a copy of <in_len> bytes plus some
+ * overhead.
  */
 static int comp_rfc195x_add_data(struct comp_ctx *comp_ctx, const char *in_data, int in_len, struct buffer *out)
 {
-       static THREAD_LOCAL struct buffer tmpbuf = BUF_NULL;
+       struct slz_stream *strm = &comp_ctx->strm;
+       int out_len;
 
        if (in_len <= 0)
                return 0;
 
-       if (comp_ctx->direct_ptr && b_is_null(&comp_ctx->queued)) {
-               /* data already being pointed to, we're in front of fragmented
-                * data and need a buffer now. We reuse the same buffer, as it's
-                * not used out of the scope of a series of add_data()*, end().
-                */
-               if (b_alloc(&tmpbuf, DB_PERMANENT) == NULL)
-                       return -1; /* no memory */
-               b_reset(&tmpbuf);
-               memcpy(b_tail(&tmpbuf), comp_ctx->direct_ptr, comp_ctx->direct_len);
-               b_add(&tmpbuf, comp_ctx->direct_len);
-               comp_ctx->direct_ptr = NULL;
-               comp_ctx->direct_len = 0;
-               comp_ctx->queued = tmpbuf;
-               /* fall through buffer copy */
-       }
+       BUG_ON(!b_room(out));
 
-       if (!b_is_null(&comp_ctx->queued)) {
-               /* data already pending */
-               memcpy(b_tail(&comp_ctx->queued), in_data, in_len);
-               b_add(&comp_ctx->queued, in_len);
-               return in_len;
-       }
-
-       comp_ctx->direct_ptr = in_data;
-       comp_ctx->direct_len = in_len;
+       out_len = slz_encode(strm, b_tail(out), in_data, in_len, 1);
+       b_add(out, out_len);
        return in_len;
 }
 
@@ -414,23 +390,10 @@ static int comp_rfc195x_add_data(struct comp_ctx *comp_ctx, const char *in_data,
 static int comp_rfc195x_flush_or_finish(struct comp_ctx *comp_ctx, struct buffer *out, int finish)
 {
        struct slz_stream *strm = &comp_ctx->strm;
-       const char *in_ptr;
-       int in_len;
        int out_len;
 
-       in_ptr = comp_ctx->direct_ptr;
-       in_len = comp_ctx->direct_len;
-
-       if (!b_is_null(&comp_ctx->queued)) {
-               in_ptr = b_head(&comp_ctx->queued);
-               in_len = b_data(&comp_ctx->queued);
-       }
-
        out_len = b_data(out);
 
-       if (in_ptr)
-               b_add(out, slz_encode(strm, b_tail(out), in_ptr, in_len, !finish));
-
        if (finish)
                b_add(out, slz_finish(strm, b_tail(out)));
        else
@@ -438,11 +401,6 @@ static int comp_rfc195x_flush_or_finish(struct comp_ctx *comp_ctx, struct buffer
 
        out_len = b_data(out) - out_len;
 
-       /* very important, we must wipe the data we've just flushed */
-       comp_ctx->direct_len = 0;
-       comp_ctx->direct_ptr = NULL;
-       comp_ctx->queued     = BUF_NULL;
-
        /* Verify compression rate limiting and CPU usage */
        if ((global.comp_rate_lim > 0 && (read_freq_ctr(&global.comp_bps_out) > global.comp_rate_lim)) ||    /* rate */
           (th_ctx->idle_pct < compress_min_idle)) {                                                         /* idle */