mirrors/haproxy
1
0
Fork 0
mirror of https://git.haproxy.org/git/haproxy.git/ synced 2025-09-21 22:01:31 +02:00
Code Issues Projects Releases Wiki Activity

MEDIUM: threads/compression: Make HTTP compression thread-safe

Browse Source
This commit is contained in:
Christopher Faulet 2017-07-25 11:07:15 +02:00 committed by Willy Tarreau
parent f227372897
commit 8ca3b4bc46
3 changed files with 57 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
include/common/hathreads.h
View File

@ -164,6 +164,7 @@ enum lock_label {
PATEXP_LOCK, PATEXP_LOCK,
PATLRU_LOCK, PATLRU_LOCK,
VARS_LOCK, VARS_LOCK,
COMP_POOL_LOCK,
LOCK_LABELS LOCK_LABELS
}; };
struct lock_stat { struct lock_stat {
@ -251,7 +252,7 @@ static inline void show_lock_stats()
"LISTENER", "LISTENER_QUEUE", "PROXY", "SERVER", "LISTENER", "LISTENER_QUEUE", "PROXY", "SERVER",
"UPDATED_SERVERS", "LBPRM", "SIGNALS", "STK_TABLE", "STK_SESS", "UPDATED_SERVERS", "LBPRM", "SIGNALS", "STK_TABLE", "STK_SESS",
"APPLETS", "PEER", "BUF_WQ", "STREAMS", "SSL", "SSL_GEN_CERTS", "APPLETS", "PEER", "BUF_WQ", "STREAMS", "SSL", "SSL_GEN_CERTS",
"PATREF", "PATEXP", "PATLRU", "VARS" }; "PATREF", "PATEXP", "PATLRU", "VARS", "COMP_POOL" };
int lbl; int lbl;
for (lbl = 0; lbl < LOCK_LABELS; lbl++) { for (lbl = 0; lbl < LOCK_LABELS; lbl++) {

66
src/compression.c
View File

@ -29,6 +29,7 @@
#include <common/cfgparse.h> #include <common/cfgparse.h>
#include <common/compat.h> #include <common/compat.h>
#include <common/memory.h> #include <common/memory.h>
#include <common/hathreads.h>
#include <types/global.h> #include <types/global.h>
#include <types/compression.h> #include <types/compression.h>
@ -40,6 +41,10 @@
#include <proto/stream.h> #include <proto/stream.h>
#ifdef USE_THREAD
static HA_SPINLOCK_T comp_pool_lock;
#endif
#ifdef USE_ZLIB #ifdef USE_ZLIB
static void *alloc_zlib(void *opaque, unsigned int items, unsigned int size); static void *alloc_zlib(void *opaque, unsigned int items, unsigned int size);
@ -154,8 +159,12 @@ static inline int init_comp_ctx(struct comp_ctx **comp_ctx)
return -1; return -1;
#endif #endif
if (unlikely(pool_comp_ctx == NULL)) if (unlikely(pool_comp_ctx == NULL)) {
pool_comp_ctx = create_pool("comp_ctx", sizeof(struct comp_ctx), MEM_F_SHARED); SPIN_LOCK(COMP_POOL_LOCK, &comp_pool_lock);
if (unlikely(pool_comp_ctx == NULL))
pool_comp_ctx = create_pool("comp_ctx", sizeof(struct comp_ctx), MEM_F_SHARED);
SPIN_UNLOCK(COMP_POOL_LOCK, &comp_pool_lock);
}
*comp_ctx = pool_alloc2(pool_comp_ctx); *comp_ctx = pool_alloc2(pool_comp_ctx);
if (*comp_ctx == NULL) if (*comp_ctx == NULL)
@ -165,7 +174,7 @@ static inline int init_comp_ctx(struct comp_ctx **comp_ctx)
(*comp_ctx)->direct_len = 0; (*comp_ctx)->direct_len = 0;
(*comp_ctx)->queued = NULL; (*comp_ctx)->queued = NULL;
#elif defined(USE_ZLIB) #elif defined(USE_ZLIB)
zlib_used_memory += sizeof(struct comp_ctx); HA_ATOMIC_ADD(&zlib_used_memory, sizeof(struct comp_ctx));
strm = &(*comp_ctx)->strm; strm = &(*comp_ctx)->strm;
strm->zalloc = alloc_zlib; strm->zalloc = alloc_zlib;
@ -187,7 +196,7 @@ static inline int deinit_comp_ctx(struct comp_ctx **comp_ctx)
*comp_ctx = NULL; *comp_ctx = NULL;
#ifdef USE_ZLIB #ifdef USE_ZLIB
zlib_used_memory -= sizeof(struct comp_ctx); HA_ATOMIC_SUB(&zlib_used_memory, sizeof(struct comp_ctx));
#endif #endif
return 0; return 0;
} }
@ -282,7 +291,7 @@ static int rfc1950_init(struct comp_ctx **comp_ctx, int level)
*/ */
static int rfc195x_add_data(struct comp_ctx *comp_ctx, const char *in_data, int in_len, struct buffer *out) static int rfc195x_add_data(struct comp_ctx *comp_ctx, const char *in_data, int in_len, struct buffer *out)
{ {
static struct buffer *tmpbuf = &buf_empty; static THREAD_LOCAL struct buffer *tmpbuf = &buf_empty;
if (in_len <= 0) if (in_len <= 0)
return 0; return 0;
@ -393,7 +402,7 @@ static int rfc195x_end(struct comp_ctx **comp_ctx)
static void *alloc_zlib(void *opaque, unsigned int items, unsigned int size) static void *alloc_zlib(void *opaque, unsigned int items, unsigned int size)
{ {
struct comp_ctx *ctx = opaque; struct comp_ctx *ctx = opaque;
static char round = 0; /* order in deflateInit2 */ static THREAD_LOCAL char round = 0; /* order in deflateInit2 */
void *buf = NULL; void *buf = NULL;
struct pool_head *pool = NULL; struct pool_head *pool = NULL;
@ -402,42 +411,62 @@ static void *alloc_zlib(void *opaque, unsigned int items, unsigned int size)
switch (round) { switch (round) {
case 0: case 0:
if (zlib_pool_deflate_state == NULL) if (zlib_pool_deflate_state == NULL) {
zlib_pool_deflate_state = create_pool("zlib_state", size * items, MEM_F_SHARED); SPIN_LOCK(COMP_POOL_LOCK, &comp_pool_lock);
if (zlib_pool_deflate_state == NULL)
zlib_pool_deflate_state = create_pool("zlib_state", size * items, MEM_F_SHARED);
SPIN_UNLOCK(COMP_POOL_LOCK, &comp_pool_lock);
}
pool = zlib_pool_deflate_state; pool = zlib_pool_deflate_state;
ctx->zlib_deflate_state = buf = pool_alloc2(pool); ctx->zlib_deflate_state = buf = pool_alloc2(pool);
break; break;
case 1: case 1:
if (zlib_pool_window == NULL) if (zlib_pool_window == NULL) {
zlib_pool_window = create_pool("zlib_window", size * items, MEM_F_SHARED); SPIN_LOCK(COMP_POOL_LOCK, &comp_pool_lock);
if (zlib_pool_window == NULL)
zlib_pool_window = create_pool("zlib_window", size * items, MEM_F_SHARED);
SPIN_UNLOCK(COMP_POOL_LOCK, &comp_pool_lock);
}
pool = zlib_pool_window; pool = zlib_pool_window;
ctx->zlib_window = buf = pool_alloc2(pool); ctx->zlib_window = buf = pool_alloc2(pool);
break; break;
case 2: case 2:
if (zlib_pool_prev == NULL) if (zlib_pool_prev == NULL) {
zlib_pool_prev = create_pool("zlib_prev", size * items, MEM_F_SHARED); SPIN_LOCK(COMP_POOL_LOCK, &comp_pool_lock);
if (zlib_pool_prev == NULL)
zlib_pool_prev = create_pool("zlib_prev", size * items, MEM_F_SHARED);
SPIN_UNLOCK(COMP_POOL_LOCK, &comp_pool_lock);
}
pool = zlib_pool_prev; pool = zlib_pool_prev;
ctx->zlib_prev = buf = pool_alloc2(pool); ctx->zlib_prev = buf = pool_alloc2(pool);
break; break;
case 3: case 3:
if (zlib_pool_head == NULL) if (zlib_pool_head == NULL) {
zlib_pool_head = create_pool("zlib_head", size * items, MEM_F_SHARED); SPIN_LOCK(COMP_POOL_LOCK, &comp_pool_lock);
if (zlib_pool_head == NULL)
zlib_pool_head = create_pool("zlib_head", size * items, MEM_F_SHARED);
SPIN_UNLOCK(COMP_POOL_LOCK, &comp_pool_lock);
}
pool = zlib_pool_head; pool = zlib_pool_head;
ctx->zlib_head = buf = pool_alloc2(pool); ctx->zlib_head = buf = pool_alloc2(pool);
break; break;
case 4: case 4:
if (zlib_pool_pending_buf == NULL) if (zlib_pool_pending_buf == NULL) {
zlib_pool_pending_buf = create_pool("zlib_pending_buf", size * items, MEM_F_SHARED); SPIN_LOCK(COMP_POOL_LOCK, &comp_pool_lock);
if (zlib_pool_pending_buf == NULL)
zlib_pool_pending_buf = create_pool("zlib_pending_buf", size * items, MEM_F_SHARED);
SPIN_UNLOCK(COMP_POOL_LOCK, &comp_pool_lock);
}
pool = zlib_pool_pending_buf; pool = zlib_pool_pending_buf;
ctx->zlib_pending_buf = buf = pool_alloc2(pool); ctx->zlib_pending_buf = buf = pool_alloc2(pool);
break; break;
} }
if (buf != NULL) if (buf != NULL)
zlib_used_memory += pool->size; HA_ATOMIC_ADD(&zlib_used_memory, pool->size);
end: end:
@ -468,7 +497,7 @@ static void free_zlib(void *opaque, void *ptr)
pool = zlib_pool_pending_buf; pool = zlib_pool_pending_buf;
pool_free2(pool, ptr); pool_free2(pool, ptr);
zlib_used_memory -= pool->size; HA_ATOMIC_SUB(&zlib_used_memory, pool->size);
} }
/************************** /**************************
@ -692,6 +721,7 @@ static void __comp_fetch_init(void)
global.tune.maxzlibmem = DEFAULT_MAXZLIBMEM * 1024U * 1024U, global.tune.maxzlibmem = DEFAULT_MAXZLIBMEM * 1024U * 1024U,
#endif #endif
#ifdef USE_ZLIB #ifdef USE_ZLIB
SPIN_INIT(&comp_pool_lock);
memprintf(&ptr, "Built with zlib version : " ZLIB_VERSION); memprintf(&ptr, "Built with zlib version : " ZLIB_VERSION);
memprintf(&ptr, "%s\nRunning on zlib version : %s", ptr, zlibVersion()); memprintf(&ptr, "%s\nRunning on zlib version : %s", ptr, zlibVersion());
#elif defined(USE_SLZ) #elif defined(USE_SLZ)

14
src/flt_http_comp.c
View File

@ -36,8 +36,8 @@ struct flt_ops comp_ops;
/* Pools used to allocate comp_state structs */ /* Pools used to allocate comp_state structs */
static struct pool_head *pool2_comp_state = NULL; static struct pool_head *pool2_comp_state = NULL;
static struct buffer *tmpbuf = &buf_empty; static THREAD_LOCAL struct buffer *tmpbuf = &buf_empty;
static struct buffer *zbuf = &buf_empty; static THREAD_LOCAL struct buffer *zbuf = &buf_empty;
struct comp_state { struct comp_state {
struct comp_ctx *comp_ctx; /* compression context */ struct comp_ctx *comp_ctx; /* compression context */
@ -66,9 +66,8 @@ static int http_compression_buffer_end(struct comp_state *st, struct stream *s,
/***********************************************************************/ /***********************************************************************/
static int static int
comp_flt_init(struct proxy *px, struct flt_conf *fconf) comp_flt_init_per_thread(struct proxy *px, struct flt_conf *fconf)
{ {
if (!tmpbuf->size && b_alloc(&tmpbuf) == NULL) if (!tmpbuf->size && b_alloc(&tmpbuf) == NULL)
return -1; return -1;
if (!zbuf->size && b_alloc(&zbuf) == NULL) if (!zbuf->size && b_alloc(&zbuf) == NULL)
@ -77,7 +76,7 @@ comp_flt_init(struct proxy *px, struct flt_conf *fconf)
} }
static void static void
comp_flt_deinit(struct proxy *px, struct flt_conf *fconf) comp_flt_deinit_per_thread(struct proxy *px, struct flt_conf *fconf)
{ {
if (tmpbuf->size) if (tmpbuf->size)
b_free(&tmpbuf); b_free(&tmpbuf);
@ -88,6 +87,7 @@ comp_flt_deinit(struct proxy *px, struct flt_conf *fconf)
static int static int
comp_start_analyze(struct stream *s, struct filter *filter, struct channel *chn) comp_start_analyze(struct stream *s, struct filter *filter, struct channel *chn)
{ {
if (filter->ctx == NULL) { if (filter->ctx == NULL) {
struct comp_state *st; struct comp_state *st;
@ -790,8 +790,8 @@ http_compression_buffer_end(struct comp_state *st, struct stream *s,
/***********************************************************************/ /***********************************************************************/
struct flt_ops comp_ops = { struct flt_ops comp_ops = {
.init = comp_flt_init, .init_per_thread = comp_flt_init_per_thread,
.deinit = comp_flt_deinit, .deinit_per_thread = comp_flt_deinit_per_thread,
.channel_start_analyze = comp_start_analyze, .channel_start_analyze = comp_start_analyze,
.channel_end_analyze = comp_end_analyze, .channel_end_analyze = comp_end_analyze,