From f028e6e066c2153442715448501b5916e2061f4b Mon Sep 17 00:00:00 2001
From: Christopher Faulet <cfaulet@haproxy.com>
Date: Wed, 22 Jun 2022 16:55:04 +0200
Subject: [PATCH] MEDIUM: bwlim: Add support of bandwith limitation at the
 stream level

This patch adds a filter to limit bandwith at the stream level. Several
filters can be defined. A filter may limit incoming data (upload) or
outgoing data (download). The limit can be defined per-stream or shared via
a stick-table. For a given stream, the bandwith limitation filters can be
enabled using the "set-bandwidth-limit" action.

A bandwith limitation filter can be used indifferently for HTTP or TCP
stream. For HTTP stream, only the payload transfer is limited. The filter is
pretty simple for now. But it was designed to be extensible. The current
design tries, as far as possible, to never exceed the limit. There is no
burst.
---
 Makefile              |   2 +-
 doc/configuration.txt | 176 +++++++++
 src/flt_bwlim.c       | 848 ++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 1025 insertions(+), 1 deletion(-)
 create mode 100644 src/flt_bwlim.c

diff --git a/Makefile b/Makefile
index b248acccfa..8148371339 100644
--- a/Makefile
+++ b/Makefile
@@ -955,7 +955,7 @@ OBJS += src/mux_h2.o src/mux_fcgi.o src/mux_h1.o src/tcpcheck.o               \
         src/base64.o src/auth.o src/uri_auth.o src/time.o src/ebistree.o      \
         src/dynbuf.o src/wdt.o src/pipe.o src/init.o src/http_acl.o           \
         src/hpack-huff.o src/hpack-enc.o src/dict.o src/freq_ctr.o            \
-        src/ebtree.o src/hash.o src/dgram.o src/version.o
+        src/ebtree.o src/hash.o src/dgram.o src/version.o src/flt_bwlim.o
 
 ifneq ($(TRACE),)
 OBJS += src/calltrace.o
diff --git a/doc/configuration.txt b/doc/configuration.txt
index 183710c357..7efefea303 100644
--- a/doc/configuration.txt
+++ b/doc/configuration.txt
@@ -120,6 +120,7 @@ Summary
 9.4.      Cache
 9.5.      fcgi-app
 9.6.      OpenTracing
+9.7.      Bandwidth limitation
 
 10.   FastCGI applications
 10.1.     Setup
@@ -6322,6 +6323,7 @@ http-request <action> [options...] [ { if | unless } <condition> ]
     - sc-inc-gpc1(<sc-id>)
     - sc-set-gpt(<idx>,<sc-id>) { <int> | <expr> }
     - sc-set-gpt0(<sc-id>) { <int> | <expr> }
+    - set-bandwidth-limit <name> [limit <expr>] [period <expr>]
     - set-dst <expr>
     - set-dst-port <expr>
     - set-header <name> <fmt>
@@ -6991,6 +6993,33 @@ http-request send-spoe-group <engine-name> <group-name>
     <group-name>   The SPOE group name as specified in the engine
                    configuration.
 
+http-request set-bandwidth-limit <name> [limit <expr>] [period <expr>] [ { if | unless } <condition> ]
+
+  This action is used to enable the bandwidth limitation filter <name>, either
+  on the upload or download direction depending on the filter type. Custom
+  limit and period may be defined, if and only if <name> references a
+  per-stream bandwidth limitation filter. When a set-bandwidth-limit rule is
+  executed, it first resets all settings of the filter to their defaults prior
+  to enabling it. As a consequence, if several "set-bandwidth-limit" actions
+  are executed for the same filter, only the last one is considered. Several
+  bandwidth limitation filters can be enabled on the same stream.
+
+  Note that this action cannot be used in a defaults section because bandwidth
+  limitation filters cannot be defined in defaults sections. In addition, only
+  the HTTP payload transfer is limited. The HTTP headers are not considered.
+
+  Arguments:
+    <expr>  Is a standard HAProxy expression formed by a sample-fetch followed
+            by some converters. The result is converted to an integer. It is
+            interpreted as a size in bytes for the "limit" parameter and as a
+            duration in milliseconds for the "period" parameter.
+
+  Example:
+    http-request set-bandwidth-limit global-limit
+    http-request set-bandwidth-limit my-limit limit 1m period 10s
+
+  See section 9.7 about bandwidth limitation filter setup.
+
 http-request set-dst <expr> [ { if | unless } <condition> ]
 
   This is used to set the destination IP address to the value of specified
@@ -7462,6 +7491,7 @@ http-response <action> <options...> [ { if | unless } <condition> ]
     - sc-set-gpt(<idx>,<sc-id>) { <int> | <expr> }
     - sc-set-gpt0(<sc-id>) { <int> | <expr> }
     - send-spoe-group <engine-name> <group-name>
+    - set-bandwidth-limit <name> [limit <expr>] [period <expr>]
     - set-header <name> <fmt>
     - set-log-level <level>
     - set-map(<file-name>) <key fmt> <value fmt>
@@ -7659,6 +7689,12 @@ http-response send-spoe-group <engine-name> <group-name>
   This action is used to trigger sending of a group of SPOE messages. Please
   refer to "http-request send-spoe-group" for a complete description.
 
+http-response set-bandwidth-limit <name> [limit <expr>] [period <expr>] [ { if | unless } <condition> ]
+
+  This action is used to enable the bandwidth limitation filter <name>, either
+  on the upload or download direction depending on the filter type. Please
+  refer to "http-request set-bandwidth-limit" for a complete description.
+
 http-response set-header <name> <fmt> [ { if | unless } <condition> ]
 
   This does the same as "http-response add-header" except that the header name
@@ -12587,6 +12623,7 @@ tcp-request content <action> [{if | unless} <condition>]
     - sc-set-gpt(<idx>,<sc-id>) { <int> | <expr> }
     - sc-set-gpt0(<sc-id>) { <int> | <expr> }
     - send-spoe-group <engine-name> <group-name>
+    - set-bandwidth-limit <name> [limit <expr>] [period <expr>]
     - set-dst <expr>
     - set-dst-port <expr>
     - set-log-level <level>
@@ -12780,6 +12817,13 @@ tcp-request content send-spoe-group <engine-name> <group-name>
   Thaction is is used to trigger sending of a group of SPOE messages. Please
   refer to "http-request send-spoe-group" for a complete description.
 
+tcp-request content set-bandwidth-limit <name> [limit <expr>] [period <expr>]
+                    [ { if | unless } <condition> ]
+
+  This action is used to enable the bandwidth limitation filter <name>, either
+  on the upload or download direction depending on the filter type. Please
+  refer to "http-request set-bandwidth-limit" for a complete description.
+
 tcp-request content set-dst <expr> [ { if | unless } <condition> ]
 tcp-request content set-dst-port <expr> [ { if | unless } <condition> ]
 
@@ -13147,6 +13191,7 @@ tcp-response content <action> [{if | unless} <condition>]
     - sc-set-gpt(<idx>,<sc-id>) { <int> | <expr> }
     - sc-set-gpt0(<sc-id>) { <int> | <expr> }
     - send-spoe-group <engine-name> <group-name>
+    - set-bandwidth-limit <name> [limit <expr>] [period <expr>]
     - set-log-level <level>
     - set-mark <mark>
     - set-nice <nice>
@@ -13224,6 +13269,14 @@ tcp-response content send-spoe-group <engine-name> <group-name>
   Thaction is is used to trigger sending of a group of SPOE messages. Please
   refer to "http-request send-spoe-group" for a complete description.
 
+
+tcp-response content set-bandwidth-limit <name> [limit <expr>] [period <expr>]
+                     [ { if | unless } <condition> ]
+
+  This action is used to enable the bandwidth limitation filter <name>, either
+  on the upload or download direction depending on the filter type. Please
+  refer to "http-request set-bandwidth-limit" for a complete description.
+
 tcp-response content set-log-level <level>  [ { if | unless } <condition> ]
 
   This action is used to set the log level of the current session. Please refer
@@ -23023,6 +23076,129 @@ filter opentracing [id <id>] config <file>
 More detailed documentation related to the operation, configuration and use
 of the filter can be found in the addons/ot directory.
 
+9.7. Bandwidth limitation
+--------------------------
+
+filter bwlim-in  <name> default-limit <size> default-period <time> [min-size <sz>]
+filter bwlim-out <name> default-limit <size> default-period <time> [min-size <sz>]
+filter bwlim-in  <name> limit <size> key <pattern> [table <table>] [min-size <sz>]
+filter bwlim-out <name> limit <size> key <pattern> [table <table>] [min-size <sz>]
+
+  Arguments :
+
+    <name>      is the filter name that will be used by 'set-bandwidth-limit'
+                actions to reference a specific bandwidth limitation filter.
+
+    <size>      is max number of bytes that can be forwarded over the period.
+                The value must be specified for per-stream and shared bandwidth
+                limitation filters. It follows the HAProxy size format and is
+                expressed in bytes.
+
+    <pattern>   is a sample expression rule as described in section 7.3. It
+                describes what elements will be analyzed, extracted, combined,
+                and used to select which table entry to update the counters. It
+                must be specified for shared bandwidth limitation filters only.
+
+    <table>     is an optional table to be used instead of the default one,
+                which is the stick-table declared in the current proxy. It can
+                be specified for shared bandwidth limitation filters only.
+
+    <time>      is the default time period used to evaluate the bandwidth
+                limitation rate. It can be speficied for per-stream bandwidth
+                limitation filters only. It follows the HAProxy time format and
+                is expressed in milliseconds.
+
+    <min-size>  is the optional minimum number of bytes forwarded at a time by
+                a stream excluding the last packet that may be smaller. This
+                value can be specified for per-stream and shared bandwidth
+                limitation filters. It follows the HAProxy size format and is
+                expressed in bytes.
+
+Bandwidth limitation filters should be used to restrict the data forwarding
+speed at the stream level. By extension, such filters limit the network
+bandwidth consumed by a resource. Several bandwidth limitation filters can be
+used. For instance, it is possible to define a limit per source address to be
+sure a client will never consume all the network bandwidth, thereby penalizing
+other clients, and another one per stream to be able to fairly handle several
+connections for a given client.
+
+The definition order of these filters is important. If several bandwidth
+filters are enabled on a stream, the filtering will be applied in their
+definition order. It is also important to understand the definition order of
+the other filters have an influence. For instance, depending on the HTTP
+compression filter is defined before or after a bandwidth limitation filter,
+the limit will be applied on the compressed payload or not. The same is true
+for the cache filter.
+
+There are two kinds of bandwidth limitation filters. The first one enforces a
+default limit and is applied per stream. The second one uses a stickiness table
+to enfore a limit equally divided between all streams sharing the same entry in
+the table.
+
+In addition, for a given filter, depending on the filter keyword used, the
+limitation can be applied on incoming data, received from the client and
+forwarded to a server, or on outgoing data, received from a server and sent to
+the client. To apply a limit on incoming data, "bwlim-in" keyword must be
+used. To apply it on outgoing data, "bwlim-out" keyword must be used. In both
+cases, the bandwidth limitation is applied on forwarded data, at the stream
+level.
+
+The bandwidth limitation is applied at the stream level and not at the
+connection level. For multiplexed protocols (H2, H3 and FastCGI), the streams
+of the same connection may have different limits.
+
+For a per-stream bandwidth limitation filter, default period and limit must be
+defined. As their names suggest, they are the default values used to setup the
+bandwidth limitation rate for a stream. However, for this kind of filter and
+only this one, it is possible to redefine these values using sample expressions
+when the filter is enabled with a TCP/HTTP "set-bandwidth-limit" action.
+
+For a shared bandwidth limitation filter, depending on whether it is applied on
+incoming or outgoing data, the stickiness table used must store the
+corresponding bytes rate information. "bytes_in_rate(<period>)" counter must be
+stored to limit incoming data and "bytes_out_rate(<period>)" counter must be
+used to limit outgoing data.
+
+Finally, it is possible to set the minimum number of bytes that a bandwidth
+limitation filter can forward at a time for a given stream. It should be used
+to not forward too small amount of data, to reduce the CPU usage. It must
+carefully be defined. Too small, a value can increase the CPU usage. Too high,
+it can increase the latency. It is also highly linked to the defined bandwidth
+limit. If it is too close to the bandwidth limit, some pauses may be
+experienced to not exceed the limit because too many bytes will be consumed at
+a time. It is highly dependent on the filter configuration. A good idea is to
+start with something around 2 TCP MSS, typically 2896 bytes, and tune it after
+some experimentations.
+
+  Example:
+    frontend http
+        bind *:80
+        mode http
+
+        # If this filter is enabled, the stream will share the download limit
+        # of 10m/s with all other streams with the same source address.
+        filter bwlim-out limit-by-src key src table limit-by-src limit 10m
+
+        # If this filter is enalbed, the stream will be limited to download at 1m/s,
+        # independently of all other streams.
+        filter bwlim-out limit-by-strm default-limit 1m default-period 1s
+
+        # Limit all streams to 1m/s (the default limit) and those accessing the
+        # internal API to 100k/s. Limit each source address to 10m/s. The shared
+        # limit is applied first. Both are limiting the download rate.
+        http-request set-bandwidth-limit limit-by-strm
+        http-request set-bandwidth-limit limit-by-strm limit 100k if { path_beg /internal }
+        http-request set-bandwidth-limit limit-by-src
+        ...
+
+    backend limit-by-src
+        # The stickiness table used by <limit-by-src> filter
+        stick-table type ip size 1m expire 3600s store bytes_out_rate(1s)
+
+See also : "tcp-request content set-bandwidth-limit",
+           "tcp-response content set-bandwidth-limit",
+           "http-request set-bandwidth-limit" and
+           "http-response set-bandwidth-limit".
 
 10. FastCGI applications
 -------------------------
diff --git a/src/flt_bwlim.c b/src/flt_bwlim.c
new file mode 100644
index 0000000000..0340e87b6f
--- /dev/null
+++ b/src/flt_bwlim.c
@@ -0,0 +1,848 @@
+/*
+ * Bandwidth limitation filter.
+ *
+ * Copyright 2022 HAProxy Technologies, Christopher Faulet <cfaulet@haproxy.com>
+ *
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version
+ * 2 of the License, or (at your option) any later version.
+ *
+ */
+
+#include <ctype.h>
+
+#include <haproxy/api.h>
+#include <haproxy/channel-t.h>
+#include <haproxy/filters.h>
+#include <haproxy/global.h>
+#include <haproxy/http_ana-t.h>
+#include <haproxy/http_rules.h>
+#include <haproxy/proxy.h>
+#include <haproxy/sample.h>
+#include <haproxy/stream.h>
+#include <haproxy/tcp_rules.h>
+#include <haproxy/time.h>
+#include <haproxy/tools.h>
+
+const char *bwlim_flt_id = "bandwidth limitation filter";
+
+struct flt_ops bwlim_ops;
+
+#define BWLIM_FL_NONE    0x00000000 /* For init purposr */
+#define BWLIM_FL_IN      0x00000001 /* Limit clients uploads */
+#define BWLIM_FL_OUT     0x00000002 /* Limit clients downloads */
+#define BWLIM_FL_SHARED  0x00000004 /* Limit shared between clients (using stick-tables) */
+
+struct bwlim_config {
+	struct proxy *proxy;
+	char         *name;
+	unsigned int flags;
+	struct sample_expr *expr;
+	union {
+		char *n;
+		struct stktable *t;
+	} table;
+	unsigned int period;
+	unsigned int limit;
+	unsigned int min_size;
+};
+
+struct bwlim_state {
+	struct freq_ctr bytes_rate;
+	struct stksess *ts;
+	struct act_rule *rule;
+	unsigned int limit;
+	unsigned int period;
+	unsigned int exp;
+};
+
+
+/* Pools used to allocate comp_state structs */
+DECLARE_STATIC_POOL(pool_head_bwlim_state, "bwlim_state", sizeof(struct bwlim_state));
+
+
+/* Apply the bandwidth limitation of the filter <filter>. <len> is the maximum
+ * amount of data that the filter can forward. This function applies the
+ * limitation and returns what the stream is authorized to forward. Several
+ * limitation can be stacked.
+ */
+static int bwlim_apply_limit(struct filter *filter, struct channel *chn, unsigned int len)
+{
+	struct bwlim_config *conf = FLT_CONF(filter);
+	struct bwlim_state *st = filter->ctx;
+	struct freq_ctr *bytes_rate;
+	unsigned int period, limit, remain, tokens, users;
+	unsigned int wait = 0;
+	int overshoot, ret = 0;
+
+	/* Don't forward anything if there is nothing to forward or the waiting
+	 * time is not expired
+	 */
+	if (!len || (tick_isset(st->exp) && !tick_is_expired(st->exp, now_ms)))
+		goto end;
+
+	st->exp = TICK_ETERNITY;
+	ret = len;
+	if (conf->flags & BWLIM_FL_SHARED) {
+		void *ptr;
+		unsigned int type = ((conf->flags & BWLIM_FL_IN) ? STKTABLE_DT_BYTES_IN_RATE : STKTABLE_DT_BYTES_OUT_RATE);
+
+		/* In shared mode, get a pointer on the stick table entry. it
+		 * will be used to get the freq-counter. It is also used to get
+		 * The number of users.
+		 */
+		ptr = stktable_data_ptr(conf->table.t, st->ts, type);
+		if (!ptr)
+			goto end;
+
+		HA_RWLOCK_WRLOCK(STK_SESS_LOCK, &st->ts->lock);
+		bytes_rate = &stktable_data_cast(ptr, std_t_frqp);
+		period = conf->table.t->data_arg[type].u;
+		limit = conf->limit;
+		users = st->ts->ref_cnt;
+	}
+	else {
+		/* On per-stream mode, the freq-counter is private to the
+		 * stream. Get it from the filter state. Rely on the custom
+		 * limit/period if defined or use the defaut ones. In this mode,
+		 * there is only one user.
+		 */
+		bytes_rate = &st->bytes_rate;
+		period = (st->period ? st->period : conf->period);
+		limit = (st->limit ? st->limit : conf->limit);
+		users = 1;
+	}
+
+	/* Be sure the current rate does not exceed the limit over the current
+	 * period. In this case, nothing is forwarded and the waiting time is
+	 * computed to be sure to not retry too early.
+	 *
+	 * The test is used to avoid the initial burst. Otherwise, streams will
+	 * consume the limit as fast as possible and will then be paused for
+	 * long time.
+	 */
+	overshoot = freq_ctr_overshoot_period(bytes_rate, period, limit);
+	if (overshoot > 0) {
+		if (conf->flags & BWLIM_FL_SHARED)
+			HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &st->ts->lock);
+		wait = div64_32((uint64_t)(conf->min_size + overshoot) * period * users,
+				limit);
+		st->exp = tick_add(now_ms, (wait ? wait : 1));
+		ret = 0;
+		goto end;
+	}
+
+	/* Get the allowed quota per user. */
+	remain = freq_ctr_remain_period(bytes_rate, period, limit, 0);
+	tokens = div64_32((uint64_t)(remain + users - 1), users);
+
+	if (tokens < len) {
+		/* The stream cannot forward all its data. But we will check if
+		 * it can perform a small burst if the global quota is large
+		 * enought. But, in this case, its waiting time will be
+		 * increased accordingly.
+		 */
+		ret = tokens;
+		if (tokens < conf->min_size) {
+			ret = (chn->flags & (CF_EOI|CF_SHUTR|CF_READ_ERROR))
+				? MIN(len, conf->min_size)
+				: conf->min_size;
+
+			if (ret <= remain)
+				wait = div64_32((uint64_t)(ret - tokens) * period * users + limit - 1, limit);
+			else
+				ret = (limit < ret) ? remain : 0;
+		}
+	}
+
+	/* At the end, update the freq-counter and compute the waiting time if
+	 * the stream is limited
+	 */
+	update_freq_ctr_period(bytes_rate, period, ret);
+	if (ret < len) {
+		wait += next_event_delay_period(bytes_rate, period, limit, MIN(len - ret, conf->min_size * users));
+		st->exp = tick_add(now_ms, (wait ? wait : 1));
+	}
+
+	if (conf->flags & BWLIM_FL_SHARED)
+		HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &st->ts->lock);
+
+  end:
+	chn->analyse_exp = tick_first((tick_is_expired(chn->analyse_exp, now_ms) ? TICK_ETERNITY : chn->analyse_exp),
+				      st->exp);
+	return ret;
+}
+
+/***************************************************************************
+ * Hooks that manage the filter lifecycle (init/check/deinit)
+ **************************************************************************/
+/* Initialize the filter. Returns -1 on error, else 0. */
+static int bwlim_init(struct proxy *px, struct flt_conf *fconf)
+{
+	fconf->flags |= FLT_CFG_FL_HTX;
+	return 0;
+}
+
+/* Free resources allocated by the bwlim filter. */
+static void bwlim_deinit(struct proxy *px, struct flt_conf *fconf)
+{
+	struct bwlim_config *conf = fconf->conf;
+
+	if (conf) {
+		free(conf->name);
+		release_sample_expr(conf->expr);
+		free(conf);
+	}
+}
+
+/* Check configuration of a bwlim filter for a specified proxy.
+ * Return 1 on error, else 0. */
+static int bwlim_check(struct proxy *px, struct flt_conf *fconf)
+{
+	struct bwlim_config *conf = fconf->conf;
+	struct stktable *target;
+
+	if (!(conf->flags & BWLIM_FL_SHARED))
+		return 0;
+
+	if (conf->table.n)
+		target = stktable_find_by_name(conf->table.n);
+	else
+		target = px->table;
+
+	if (!target) {
+		ha_alert("Proxy %s : unable to find table '%s' referenced by bwlim filter '%s'",
+			 px->id, conf->table.n ? conf->table.n : px->id, conf->name);
+		return 1;
+	}
+
+	if ((conf->flags & BWLIM_FL_IN) && !target->data_ofs[STKTABLE_DT_BYTES_IN_RATE]) {
+		ha_alert("Proxy %s : stick-table '%s' uses a data type incompatible with bwlim filter '%s'."
+			 " It must be 'bytes_in_rate'",
+			 px->id, conf->table.n ? conf->table.n : px->id, conf->name);
+		return 1;
+	}
+	else if ((conf->flags & BWLIM_FL_OUT) && !target->data_ofs[STKTABLE_DT_BYTES_OUT_RATE]) {
+		ha_alert("Proxy %s : stick-table '%s' uses a data type incompatible with bwlim filter '%s'."
+			 " It must be 'bytes_out_rate'",
+			 px->id, conf->table.n ? conf->table.n : px->id, conf->name);
+		return 1;
+	}
+
+	if (!stktable_compatible_sample(conf->expr,  target->type)) {
+		ha_alert("Proxy %s : stick-table '%s' uses a key type incompatible with bwlim filter '%s'",
+			 px->id, conf->table.n ? conf->table.n : px->id, conf->name);
+		return 1;
+	}
+	else {
+		if (!in_proxies_list(target->proxies_list, px)) {
+			px->next_stkt_ref = target->proxies_list;
+			target->proxies_list = px;
+		}
+		free(conf->table.n);
+		conf->table.t = target;
+	}
+
+	return 0;
+}
+
+/**************************************************************************
+ * Hooks to handle start/stop of streams
+ *************************************************************************/
+/* Called when a filter instance is created and attach to a stream */
+static int bwlim_attach(struct stream *s, struct filter *filter)
+{
+	struct bwlim_state *st;
+
+	st = pool_zalloc(pool_head_bwlim_state);
+	if (!st)
+		return -1;
+	filter->ctx = st;
+	return 1;
+}
+
+/* Called when a filter instance is detach from a stream, just before its
+ * destruction */
+static void bwlim_detach(struct stream *s, struct filter *filter)
+{
+	struct bwlim_config *conf = FLT_CONF(filter);
+	struct bwlim_state *st = filter->ctx;
+	struct stktable *t = conf->table.t;
+
+	if (!st)
+		return;
+
+	if (st->ts)
+		stktable_touch_local(t, st->ts, 1);
+
+	/* release any possible compression context */
+	pool_free(pool_head_bwlim_state, st);
+	filter->ctx = NULL;
+}
+
+/**************************************************************************
+ * Hooks to filter HTTP messages
+ *************************************************************************/
+static int bwlim_http_headers(struct stream *s, struct filter *filter, struct http_msg *msg)
+{
+	msg->chn->analyse_exp = TICK_ETERNITY;
+	return 1;
+}
+
+static int bwlim_http_payload(struct stream *s, struct filter *filter, struct http_msg *msg,
+			      unsigned int offset, unsigned int len)
+{
+	return bwlim_apply_limit(filter, msg->chn, len);
+}
+
+/**************************************************************************
+ * Hooks to filter TCP data
+ *************************************************************************/
+static int bwlim_tcp_payload(struct stream *s, struct filter *filter, struct channel *chn,
+			     unsigned int offset, unsigned int len)
+{
+	return bwlim_apply_limit(filter, chn, len);
+}
+
+/********************************************************************
+ * Functions that manage the filter initialization
+ ********************************************************************/
+struct flt_ops bwlim_ops = {
+	/* Manage bwlim filter, called for each filter declaration */
+	.init              = bwlim_init,
+	.deinit            = bwlim_deinit,
+	.check             = bwlim_check,
+
+	/* Handle start/stop of streams */
+	.attach             = bwlim_attach,
+	.detach             = bwlim_detach,
+
+
+	/* Filter HTTP requests and responses */
+	.http_headers        = bwlim_http_headers,
+	.http_payload        = bwlim_http_payload,
+
+	/* Filter TCP data */
+	.tcp_payload        = bwlim_tcp_payload,
+};
+
+/* Set a bandwidth limitation. It always return ACT_RET_CONT. On error, the rule
+ * is ignored. First of all, it looks for the corresponding filter. Then, for a
+ * shared limitation, the stick-table entry is retrieved. For a per-stream
+ * limitation, the custom limit and period are computed, if necessary. At the
+ * end, the filter is registered on the data filtering for the right channel
+ * (bwlim-in = request, bwlim-out = response).
+ */
+static enum act_return bwlim_set_limit(struct act_rule *rule, struct proxy *px,
+				       struct session *sess, struct stream *s, int flags)
+{
+	struct bwlim_config *conf = rule->arg.act.p[3];
+	struct filter *filter;
+	struct bwlim_state *st = NULL;
+	struct stktable *t;
+	struct stktable_key *key;
+	struct stksess *ts;
+	int opt;
+
+	list_for_each_entry(filter, &s->strm_flt.filters, list) {
+		if (FLT_ID(filter) == bwlim_flt_id && FLT_CONF(filter) == conf) {
+			st = filter->ctx;
+			break;
+		}
+	}
+
+	if (!st)
+		goto end;
+
+	switch (rule->from) {
+	case ACT_F_TCP_REQ_CNT: opt = SMP_OPT_DIR_REQ | SMP_OPT_FINAL; break;
+	case ACT_F_TCP_RES_CNT: opt = SMP_OPT_DIR_RES | SMP_OPT_FINAL; break;
+	case ACT_F_HTTP_REQ:    opt = SMP_OPT_DIR_REQ | SMP_OPT_FINAL; break;
+	case ACT_F_HTTP_RES:    opt = SMP_OPT_DIR_RES | SMP_OPT_FINAL; break;
+	default:
+		goto end;
+	}
+
+	if (conf->flags & BWLIM_FL_SHARED) {
+		t = conf->table.t;
+		key = stktable_fetch_key(t, px, sess, s, opt, conf->expr, NULL);
+		if (!key)
+			goto end;
+
+		ts = stktable_get_entry(t, key);
+		if (!ts)
+			goto end;
+
+		st->ts = ts;
+		st->rule = rule;
+	}
+	else {
+		struct sample *smp;
+
+		st->limit = 0;
+		st->period = 0;
+		if (rule->arg.act.p[1]) {
+			smp = sample_fetch_as_type(px, sess, s, opt, rule->arg.act.p[1], SMP_T_SINT);
+			if (smp && smp->data.u.sint > 0)
+				st->limit = smp->data.u.sint;
+		}
+		if (rule->arg.act.p[2]) {
+			smp = sample_fetch_as_type(px, sess, s, opt, rule->arg.act.p[2], SMP_T_SINT);
+			if (smp && smp->data.u.sint > 0)
+				st->period = smp->data.u.sint;
+		}
+	}
+
+	st->exp = TICK_ETERNITY;
+	if (conf->flags & BWLIM_FL_IN)
+		register_data_filter(s, &s->req, filter);
+	else
+		register_data_filter(s, &s->res, filter);
+
+  end:
+	return ACT_RET_CONT;
+}
+
+/* Check function for "set-bandwidth-limit" aciton. It returns 1 on
+ * success. Otherwise, it returns 0 and <err> is filled.
+ */
+int check_bwlim_action(struct act_rule *rule, struct proxy *px, char **err)
+{
+	struct flt_conf *fconf;
+	struct bwlim_config *conf = NULL;
+	unsigned int where;
+
+	list_for_each_entry(fconf, &px->filter_configs, list) {
+		conf = NULL;
+		if (fconf->id == bwlim_flt_id) {
+			conf = fconf->conf;
+			if (!strcmp(rule->arg.act.p[0], conf->name))
+				break;
+		}
+	}
+	if (!conf) {
+		memprintf(err, "unable to find bwlim filter '%s' referenced by set-bandwidth-limit rule",
+			  (char *)rule->arg.act.p[0]);
+		return 0;
+	}
+
+	if ((conf->flags & BWLIM_FL_SHARED) && rule->arg.act.p[1]) {
+		memprintf(err, "set-bandwidth-limit rule cannot define a limit for a shared bwlim filter");
+		return 0;
+	}
+
+	where = 0;
+	if (px->cap & PR_CAP_FE)
+		where |= (rule->from == ACT_F_HTTP_REQ ? SMP_VAL_FE_HRQ_HDR : SMP_VAL_FE_HRS_HDR);
+	if (px->cap & PR_CAP_BE)
+		where |= (rule->from == ACT_F_HTTP_REQ ? SMP_VAL_BE_HRQ_HDR : SMP_VAL_BE_HRS_HDR);
+
+	if (rule->arg.act.p[1]) {
+		struct sample_expr *expr = rule->arg.act.p[1];
+
+		if (!(expr->fetch->val & where)) {
+			memprintf(err, "set-bandwidth-limit rule uses a limit extracting information from '%s', none of which is available here",
+				  sample_src_names(expr->fetch->use));
+			return 0;
+		}
+
+		if (rule->from == ACT_F_TCP_REQ_CNT && (px->cap & PR_CAP_FE)) {
+			if (!px->tcp_req.inspect_delay && !(expr->fetch->val & SMP_VAL_FE_SES_ACC)) {
+				ha_warning("%s '%s' : a 'tcp-request content set-bandwidth-limit*' rule explicitly depending on request"
+					   " contents without any 'tcp-request inspect-delay' setting."
+					   " This means that this rule will randomly find its contents. This can be fixed by"
+					   " setting the tcp-request inspect-delay.\n",
+					   proxy_type_str(px), px->id);
+			}
+		}
+	}
+
+	if (conf->expr) {
+		if (!(conf->expr->fetch->val & where)) {
+			memprintf(err, "bwlim filter '%s uses a key extracting information from '%s', none of which is available here",
+				  conf->name, sample_src_names(conf->expr->fetch->use));
+			return 0;
+		}
+
+		if (rule->from == ACT_F_TCP_REQ_CNT && (px->cap & PR_CAP_FE)) {
+			if (!px->tcp_req.inspect_delay && !(conf->expr->fetch->val & SMP_VAL_FE_SES_ACC)) {
+				ha_warning("%s '%s' : a 'tcp-request content set-bandwidth-limit*' rule explicitly depending on request"
+					   " contents without any 'tcp-request inspect-delay' setting."
+					   " This means that this rule will randomly find its contents. This can be fixed by"
+					   " setting the tcp-request inspect-delay.\n",
+					   proxy_type_str(px), px->id);
+			}
+		}
+	}
+
+  end:
+	rule->arg.act.p[3] = conf;
+	return 1;
+}
+
+/* Release memory allocated by "set-bandwidth-limit" action. */
+static void release_bwlim_action(struct act_rule *rule)
+{
+	free(rule->arg.act.p[0]);
+	if (rule->arg.act.p[1])
+	    release_sample_expr(rule->arg.act.p[1]);
+	if (rule->arg.act.p[2])
+	    release_sample_expr(rule->arg.act.p[2]);
+}
+
+/* Parse "set-bandwidth-limit" action. The filter name must be specified. For
+ * shared limitations, there is no other supported parameter. For per-stream
+ * limitations, a custom limit and period may be specified. In both case, it
+ * must be an expression. On success:
+ *
+ *   arg.act.p[0] will be the filter name (mandatory)
+ *   arg.act.p[1] will be an expression for the custom limit (optional, may be NULL)
+ *   arg.act.p[2] will be an expression for the custom period (optiona, may be NULLl)
+ *
+ * It returns ACT_RET_PRS_OK on success, ACT_RET_PRS_ERR on error.
+ */
+static enum act_parse_ret parse_bandwidth_limit(const char **args, int *orig_arg, struct proxy *px,
+						struct act_rule *rule, char **err)
+{
+	struct sample_expr *expr;
+	int cur_arg;
+
+	cur_arg = *orig_arg;
+
+	if (!*args[cur_arg]) {
+		memprintf(err, "missing bwlim filter name");
+		return ACT_RET_PRS_ERR;
+	}
+
+	rule->arg.act.p[0] = strdup(args[cur_arg]);
+	if (!rule->arg.act.p[0]) {
+		memprintf(err, "out of memory");
+		return ACT_RET_PRS_ERR;
+	}
+	cur_arg++;
+
+	while (1) {
+		if (strcmp(args[cur_arg], "limit") == 0) {
+			cur_arg++;
+			if (!args[cur_arg]) {
+				memprintf(err, "missing limit expression");
+				goto error;
+			}
+
+			expr = sample_parse_expr((char **)args, &cur_arg, px->conf.args.file, px->conf.args.line, err, &px->conf.args, NULL);
+			if (!expr)
+				goto error;
+			rule->arg.act.p[1] = expr;
+		}
+		else if (strcmp(args[cur_arg], "period") == 0) {
+			cur_arg++;
+			if (!args[cur_arg]) {
+				memprintf(err, "missing period expression");
+				goto error;
+			}
+
+			expr = sample_parse_expr((char **)args, &cur_arg, px->conf.args.file, px->conf.args.line, err, &px->conf.args, NULL);
+			if (!expr)
+				goto error;
+			rule->arg.act.p[2] = expr;
+		}
+		else
+			break;
+	}
+
+	rule->action_ptr = bwlim_set_limit;
+	rule->check_ptr = check_bwlim_action;
+	rule->release_ptr = release_bwlim_action;
+
+	*orig_arg = cur_arg;
+	return ACT_RET_PRS_OK;
+
+error:
+	release_bwlim_action(rule);
+	return ACT_RET_PRS_ERR;
+}
+
+
+static struct action_kw_list tcp_req_cont_actions = {
+	.kw = {
+		{ "set-bandwidth-limit", parse_bandwidth_limit, 0 },
+		{ NULL, NULL }
+	}
+};
+
+static struct action_kw_list tcp_res_cont_actions = {
+	.kw = {
+		{ "set-bandwidth-limit", parse_bandwidth_limit, 0 },
+		{ NULL, NULL }
+	}
+};
+
+static struct action_kw_list http_req_actions = {
+	.kw = {
+		{ "set-bandwidth-limit", parse_bandwidth_limit, 0 },
+		{ NULL, NULL }
+	}
+};
+
+static struct action_kw_list http_res_actions = {
+	.kw = {
+		{ "set-bandwidth-limit", parse_bandwidth_limit, 0 },
+		{ NULL, NULL }
+	}
+};
+
+INITCALL1(STG_REGISTER, tcp_req_cont_keywords_register, &tcp_req_cont_actions);
+INITCALL1(STG_REGISTER, tcp_res_cont_keywords_register, &tcp_res_cont_actions);
+INITCALL1(STG_REGISTER, http_req_keywords_register, &http_req_actions);
+INITCALL1(STG_REGISTER, http_res_keywords_register, &http_res_actions);
+
+
+/* Generic function to parse bandwidth limitation filter configurartion. It
+ * Returns -1 on error and 0 on success. It handles configuration for per-stream
+ * and shared limitations.
+ */
+static int parse_bwlim_flt(char **args, int *cur_arg, struct proxy *px, struct flt_conf *fconf,
+			   char **err, void *private)
+{
+	struct flt_conf *fc;
+	struct bwlim_config *conf;
+	int shared, per_stream;
+	int pos = *cur_arg + 1;
+
+	conf = calloc(1, sizeof(*conf));
+	if (!conf) {
+		memprintf(err, "%s: out of memory", args[*cur_arg]);
+		return -1;
+	}
+	conf->proxy = px;
+
+	if (!*args[pos]) {
+		memprintf(err, "'%s' : a name is expected as first argument ", args[*cur_arg]);
+		goto error;
+	}
+	conf->flags = BWLIM_FL_NONE;
+	conf->name = strdup(args[pos]);
+	if (!conf->name) {
+		memprintf(err, "%s: out of memory", args[*cur_arg]);
+		goto error;
+	}
+
+	list_for_each_entry(fc, &px->filter_configs, list) {
+		if (fc->id == bwlim_flt_id) {
+			struct bwlim_config *c = fc->conf;
+
+			if (!strcmp(conf->name, c->name)) {
+				memprintf(err, "bwlim filter '%s' already declared for proxy '%s'\n",
+					  conf->name, px->id);
+				goto error;
+			}
+		}
+	}
+	shared = per_stream = 0;
+	pos++;
+	while (*args[pos]) {
+		if (strcmp(args[pos], "key") == 0) {
+			if (per_stream) {
+				memprintf(err, "'%s' : cannot mix per-stream and shared parameter",
+					  args[*cur_arg]);
+				goto error;
+			}
+			if (!*args[pos + 1]) {
+				memprintf(err, "'%s' : the sample expression is missing for '%s' option",
+					  args[*cur_arg], args[pos]);
+				goto error;
+			}
+			shared = 1;
+			pos++;
+			conf->expr = sample_parse_expr((char **)args, &pos, px->conf.args.file, px->conf.args.line,
+						       err, &px->conf.args, NULL);
+			if (!conf->expr)
+				goto error;
+		}
+		else if (strcmp(args[pos], "table") == 0) {
+			if (per_stream) {
+				memprintf(err, "'%s' : cannot mix per-stream and shared parameter",
+					  args[*cur_arg]);
+				goto error;
+			}
+			if (!*args[pos + 1]) {
+				memprintf(err, "'%s' : the table name is missing for '%s' option",
+					  args[*cur_arg], args[pos]);
+				goto error;
+			}
+			shared = 1;
+			conf->table.n = strdup(args[pos + 1]);
+			if (!conf->table.n) {
+				memprintf(err, "%s: out of memory", args[*cur_arg]);
+				goto error;
+			}
+			pos += 2;
+		}
+		else if (strcmp(args[pos], "default-period") == 0) {
+			const char *res;
+
+			if (shared) {
+				memprintf(err, "'%s' : cannot mix per-stream and shared parameter",
+					  args[*cur_arg]);
+				goto error;
+			}
+			if (!*args[pos + 1]) {
+				memprintf(err, "'%s' : the value is missing for '%s' option",
+					  args[*cur_arg], args[pos]);
+				goto error;
+			}
+			per_stream = 1;
+			res = parse_time_err(args[pos + 1], &conf->period, TIME_UNIT_MS);
+                        if (res) {
+                                memprintf(err, "'%s' : invalid value for option '%s' (unexpected character '%c')",
+					  args[*cur_arg], args[pos], *res);
+				goto error;
+                        }
+			pos += 2;
+		}
+		else if (strcmp(args[pos], "limit") == 0) {
+			const char *res;
+
+			if (per_stream) {
+				memprintf(err, "'%s' : cannot mix per-stream and shared parameter",
+					  args[*cur_arg]);
+				goto error;
+			}
+			if (!*args[pos + 1]) {
+				memprintf(err, "'%s' : the value is missing for '%s' option",
+					  args[*cur_arg], args[pos]);
+				goto error;
+			}
+			shared = 1;
+			res = parse_size_err(args[pos + 1], &conf->limit);
+                        if (res) {
+                                memprintf(err, "'%s' : invalid value for option '%s' (unexpected character '%c')",
+					  args[*cur_arg], args[pos], *res);
+				goto error;
+                        }
+			pos += 2;
+		}
+		else if (strcmp(args[pos], "default-limit") == 0) {
+			const char *res;
+
+			if (shared) {
+				memprintf(err, "'%s' : cannot mix per-stream and shared parameter",
+					  args[*cur_arg]);
+				goto error;
+			}
+			if (!*args[pos + 1]) {
+				memprintf(err, "'%s' : the value is missing for '%s' option",
+					  args[*cur_arg], args[pos]);
+				goto error;
+			}
+			per_stream = 1;
+			res = parse_size_err(args[pos + 1], &conf->limit);
+                        if (res) {
+                                memprintf(err, "'%s' : invalid value for option '%s' (unexpected character '%c')",
+					  args[*cur_arg], args[pos], *res);
+				goto error;
+                        }
+			pos += 2;
+		}
+		else if (strcmp(args[pos], "min-size") == 0) {
+			const char *res;
+
+			if (!*args[pos + 1]) {
+				memprintf(err, "'%s' : the value is missing for '%s' option",
+					  args[*cur_arg], args[pos]);
+				goto error;
+			}
+			res = parse_size_err(args[pos + 1], &conf->min_size);
+                        if (res) {
+                                memprintf(err, "'%s' : invalid value for option '%s' (unexpected character '%c')",
+					  args[*cur_arg], args[pos], *res);
+				goto error;
+                        }
+			pos += 2;
+		}
+		else
+			break;
+	}
+
+	if (shared) {
+		conf->flags |= BWLIM_FL_SHARED;
+		if (!conf->expr) {
+			memprintf(err, "'%s' : <key> option is missing", args[*cur_arg]);
+			goto error;
+		}
+		if (!conf->limit) {
+			memprintf(err, "'%s' : <limit> option is missing", args[*cur_arg]);
+			goto error;
+		}
+	}
+	else {
+		/* Per-stream: limit downloads only for now */
+		conf->flags |= BWLIM_FL_OUT;
+		if (!conf->period) {
+			memprintf(err, "'%s' : <default-period> option is missing", args[*cur_arg]);
+			goto error;
+		}
+		if (!conf->limit) {
+			memprintf(err, "'%s' : <default-limit> option is missing", args[*cur_arg]);
+			goto error;
+		}
+	}
+
+	*cur_arg = pos;
+	fconf->id   = bwlim_flt_id;
+	fconf->ops  = &bwlim_ops;
+	fconf->conf = conf;
+	return 0;
+
+ error:
+	if (conf->name)
+		free(conf->name);
+	if (conf->expr)
+		release_sample_expr(conf->expr);
+	if (conf->table.n)
+		free(conf->table.n);
+	free(conf);
+	return -1;
+}
+
+
+static int parse_bwlim_in_flt(char **args, int *cur_arg, struct proxy *px, struct flt_conf *fconf,
+			      char **err, void *private)
+{
+	int ret;
+
+	ret = parse_bwlim_flt(args, cur_arg, px, fconf, err, private);
+	if (!ret) {
+		struct bwlim_config *conf = fconf->conf;
+
+		conf->flags |= BWLIM_FL_IN;
+	}
+
+	return ret;
+}
+
+static int parse_bwlim_out_flt(char **args, int *cur_arg, struct proxy *px, struct flt_conf *fconf,
+			       char **err, void *private)
+{
+	int ret;
+
+	ret = parse_bwlim_flt(args, cur_arg, px, fconf, err, private);
+	if (!ret) {
+		struct bwlim_config *conf = fconf->conf;
+
+		conf->flags |= BWLIM_FL_OUT;
+	}
+	return ret;
+}
+
+/* Declare the filter parser for "trace" keyword */
+static struct flt_kw_list flt_kws = { "BWLIM", { }, {
+		{ "bwlim-in",  parse_bwlim_in_flt, NULL },
+		{ "bwlim-out", parse_bwlim_out_flt, NULL },
+		{ NULL, NULL, NULL },
+	}
+};
+
+INITCALL1(STG_REGISTER, flt_register_keywords, &flt_kws);