Changed limit_rate algorithm to leaky bucket.

The bucket size (burst size) is set to the size expected to be sent
in 1 second (that is, the rate configured), so the first operation will
be able to send corresponding number of bytes, matching the previous
behaviour.  To ensure that inaccurate timers won't affect rate limiting
accuracy, delays are set to trigger next sending when the bucket is
half-empty.

The limit_rate_after directive makes the bucket size larger, allowing
for larger traffic bursts (it won't affect delays though).  This is mostly
equivalent to the previous behaviour in simple cases, like downloading
a static file by a fast client, but expected to work better in complex
cases, such as when actual response traffic might stop for a while.

Similar changes are made to proxy_limit_rate (and friends) in the http
module, as well as proxy_upload_rate and proxy_download_rate in the stream
module.

Immediate benefits include more accurate limiting, notably at rates which
resulted in 1 or 2 millisecond delays after sending (and very inaccurate
limiting) with the old algorithm.  Further, there will be less unneeded
delays, notably no delays at all in most cases as long as the client is
not using allowed bandwidth.

Note that due to the new algorithm there are minor changes in the limiting,
mostly visible with very small limits: typically 1.5x of the limit will
be sent in the first second (one immediately, and half after 0.5 seconds
delay), and additional data will be sent once per 0.5 seconds (instead
of once per second previously).  Some tests needs to be adapted for this.

Note that r->limit_last is not initialized, and will be set to an actual
value only after first sending.  As a result, "ms" can be large during
initial sending, and to ensure that there will be no signed integer
overflows, calculations of "excess" are performed in (uint64_t) (and the
result is type casted into (off_t) to clarify that the code intentionally
relies on implementation-defined behaviour, similarly to how we handle
calculations of "ms").

Author: mdounin

src/event/ngx_event_pipe.c

@@ -108,12 +108,13 @@ ngx_event_pipe(ngx_event_pipe_t *p, ngx_int_t do_write)
 static ngx_int_t
 ngx_event_pipe_read_upstream(ngx_event_pipe_t *p)
 {
-    off_t         limit;
-    ssize_t       n, size;
-    ngx_int_t     rc;
-    ngx_buf_t    *b;
-    ngx_msec_t    delay;
-    ngx_chain_t  *chain, *cl, *ln;
+    off_t            limit, excess;
+    ssize_t          n, size, sent;
+    ngx_int_t        rc;
+    ngx_buf_t       *b;
+    ngx_msec_t       delay;
+    ngx_chain_t     *chain, *cl, *ln;
+    ngx_msec_int_t   ms;
 
     if (p->upstream_eof || p->upstream_error || p->upstream_done
         || p->upstream == NULL)
@@ -145,6 +146,8 @@ ngx_event_pipe_read_upstream(ngx_event_pipe_t *p)
     ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0,
                    "pipe read upstream: %d", p->upstream->read->ready);
 
+    excess = 0;
+
     for ( ;; ) {
 
         if (p->upstream_eof || p->upstream_error || p->upstream_done) {
@@ -212,12 +215,19 @@ ngx_event_pipe_read_upstream(ngx_event_pipe_t *p)
                     break;
                 }
 
-                limit = (off_t) p->limit_rate * (ngx_time() - p->start_sec + 1)
-                        - p->read_length;
+                ms = (ngx_msec_int_t) (ngx_current_msec - p->limit_last);
+                ms = ngx_max(ms, 0);
+
+                excess = (off_t) (p->limit_excess
+                                  - (uint64_t) p->limit_rate * ms / 1000);
+                excess = ngx_max(excess, 0);
+
+                limit = (off_t) p->limit_rate - excess;
 
                 if (limit <= 0) {
                     p->upstream->read->delayed = 1;
-                    delay = (ngx_msec_t) (- limit * 1000 / p->limit_rate + 1);
+                    excess -= (off_t) p->limit_rate / 2;
+                    delay = (ngx_msec_t) (excess * 1000 / p->limit_rate + 1);
                     ngx_add_timer(p->upstream->read, delay);
                     break;
                 }
@@ -345,8 +355,7 @@ ngx_event_pipe_read_upstream(ngx_event_pipe_t *p)
             }
         }
 
-        delay = p->limit_rate ? (ngx_msec_t) n * 1000 / p->limit_rate : 0;
-
+        sent = n;
         p->read_length += n;
         cl = chain;
         p->free_raw_bufs = NULL;
@@ -384,10 +393,22 @@ ngx_event_pipe_read_upstream(ngx_event_pipe_t *p)
             p->free_raw_bufs = cl;
         }
 
-        if (delay > 0) {
-            p->upstream->read->delayed = 1;
-            ngx_add_timer(p->upstream->read, delay);
-            break;
+        if (p->limit_rate) {
+            excess += sent;
+
+            p->limit_last = ngx_current_msec;
+            p->limit_excess = excess;
+
+            excess -= (off_t) p->limit_rate / 2;
+            excess = ngx_max(excess, 0);
+
+            delay = (ngx_msec_t) (excess * 1000 / p->limit_rate);
+
+            if (delay > 0) {
+                p->upstream->read->delayed = 1;
+                ngx_add_timer(p->upstream->read, delay);
+                break;
+            }
         }
     }
 

src/event/ngx_event_pipe.h

@@ -91,7 +91,8 @@ struct ngx_event_pipe_s {
     ngx_buf_t         *buf_to_file;
 
     size_t             limit_rate;
-    time_t             start_sec;
+    ngx_msec_t         limit_last;
+    off_t              limit_excess;
 
     ngx_temp_file_t   *temp_file;
 

src/http/ngx_http_request.h

@@ -448,6 +448,9 @@ struct ngx_http_request_s {
     size_t                            limit_rate;
     size_t                            limit_rate_after;
 
+    ngx_msec_t                        limit_last;
+    off_t                             limit_excess;
+
     /* used to learn the Apache compatible response length without a header */
     size_t                            header_size;
 

src/http/ngx_http_upstream.c

@@ -3271,7 +3271,6 @@ ngx_http_upstream_send_response(ngx_http_request_t *r, ngx_http_upstream_t *u)
     p->pool = r->pool;
     p->log = c->log;
     p->limit_rate = u->conf->limit_rate;
-    p->start_sec = ngx_time();
 
     p->cacheable = u->cacheable || u->store;
 

src/http/ngx_http_write_filter_module.c

@@ -47,10 +47,11 @@ ngx_module_t  ngx_http_write_filter_module = {
 ngx_int_t
 ngx_http_write_filter(ngx_http_request_t *r, ngx_chain_t *in)
 {
-    off_t                      size, sent, nsent, limit;
+    off_t                      size, sent, excess, limit;
     ngx_uint_t                 last, flush, sync;
     ngx_msec_t                 delay;
     ngx_chain_t               *cl, *ln, **ll, *chain;
+    ngx_msec_int_t             ms;
     ngx_connection_t          *c;
     ngx_http_core_loc_conf_t  *clcf;
 
@@ -66,6 +67,10 @@ ngx_http_write_filter(ngx_http_request_t *r, ngx_chain_t *in)
     last = 0;
     ll = &r->out;
 
+#if (NGX_SUPPRESS_WARN)
+    excess = 0;
+#endif
+
     /* find the size, the flush point and the last link of the saved chain */
 
     for (cl = r->out; cl; cl = cl->next) {
@@ -268,12 +273,19 @@ ngx_http_write_filter(ngx_http_request_t *r, ngx_chain_t *in)
             r->limit_rate_after_set = 1;
         }
 
-        limit = (off_t) r->limit_rate * (ngx_time() - r->start_sec + 1)
-                - (c->sent - r->limit_rate_after);
+        ms = (ngx_msec_int_t) (ngx_current_msec - r->limit_last);
+        ms = ngx_max(ms, 0);
+
+        excess = (off_t) (r->limit_excess
+                          - (uint64_t) r->limit_rate * ms / 1000);
+        excess = ngx_max(excess, 0);
+
+        limit = (off_t) r->limit_rate + (off_t) r->limit_rate_after - excess;
 
         if (limit <= 0) {
             c->write->delayed = 1;
-            delay = (ngx_msec_t) (- limit * 1000 / r->limit_rate + 1);
+            excess -= (off_t) r->limit_rate_after + (off_t) r->limit_rate / 2;
+            delay = (ngx_msec_t) (excess * 1000 / r->limit_rate + 1);
             ngx_add_timer(c->write, delay);
 
             c->buffered |= NGX_HTTP_WRITE_BUFFERED;
@@ -308,22 +320,15 @@ ngx_http_write_filter(ngx_http_request_t *r, ngx_chain_t *in)
 
     if (r->limit_rate) {
 
-        nsent = c->sent;
+        excess += (c->sent - sent);
 
-        if (r->limit_rate_after) {
+        r->limit_last = ngx_current_msec;
+        r->limit_excess = excess;
 
-            sent -= r->limit_rate_after;
-            if (sent < 0) {
-                sent = 0;
-            }
-
-            nsent -= r->limit_rate_after;
-            if (nsent < 0) {
-                nsent = 0;
-            }
-        }
+        excess -= (off_t) r->limit_rate_after + (off_t) r->limit_rate / 2;
+        excess = ngx_max(excess, 0);
 
-        delay = (ngx_msec_t) ((nsent - sent) * 1000 / r->limit_rate);
+        delay = (ngx_msec_t) (excess * 1000 / r->limit_rate);
 
         if (delay > 0) {
             c->write->delayed = 1;

src/stream/ngx_stream_proxy_module.c

@@ -435,7 +435,6 @@ ngx_stream_proxy_handler(ngx_stream_session_t *s)
     }
 
     u->peer.type = c->type;
-    u->start_sec = ngx_time();
 
     c->write->handler = ngx_stream_proxy_downstream_handler;
     c->read->handler = ngx_stream_proxy_downstream_handler;
@@ -924,6 +923,9 @@ ngx_stream_proxy_init_upstream(ngx_stream_session_t *s)
     u->upload_rate = ngx_stream_complex_value_size(s, pscf->upload_rate, 0);
     u->download_rate = ngx_stream_complex_value_size(s, pscf->download_rate, 0);
 
+    u->upload_last = ngx_current_msec;
+    u->upload_excess = s->received;
+
     u->connected = 1;
 
     pc->read->handler = ngx_stream_proxy_upstream_handler;
@@ -1555,14 +1557,15 @@ ngx_stream_proxy_process(ngx_stream_session_t *s, ngx_uint_t from_upstream,
     ngx_uint_t do_write)
 {
     char                         *recv_action, *send_action;
-    off_t                        *received, limit, sent;
+    off_t                        *received, *limit_excess, limit, excess, sent;
     size_t                        size, limit_rate;
     ssize_t                       n;
     ngx_buf_t                    *b;
     ngx_int_t                     rc;
     ngx_uint_t                    flags, *packets;
-    ngx_msec_t                    delay;
+    ngx_msec_t                   *limit_last, delay;
     ngx_chain_t                  *cl, **ll, **out, **busy;
+    ngx_msec_int_t                ms;
     ngx_connection_t             *c, *pc, *src, *dst;
     ngx_log_handler_pt            handler;
     ngx_stream_upstream_t        *u;
@@ -1595,6 +1598,8 @@ ngx_stream_proxy_process(ngx_stream_session_t *s, ngx_uint_t from_upstream,
         dst = c;
         b = &u->upstream_buf;
         limit_rate = u->download_rate;
+        limit_last = &u->download_last;
+        limit_excess = &u->download_excess;
         received = &u->received;
         packets = &u->responses;
         out = &u->downstream_out;
@@ -1607,6 +1612,8 @@ ngx_stream_proxy_process(ngx_stream_session_t *s, ngx_uint_t from_upstream,
         dst = pc;
         b = &u->downstream_buf;
         limit_rate = u->upload_rate;
+        limit_last = &u->upload_last;
+        limit_excess = &u->upload_excess;
         received = &s->received;
         packets = &u->requests;
         out = &u->upstream_out;
@@ -1616,6 +1623,7 @@ ngx_stream_proxy_process(ngx_stream_session_t *s, ngx_uint_t from_upstream,
     }
 
 #if (NGX_SUPPRESS_WARN)
+    excess = 0;
     sent = 0;
 #endif
 
@@ -1652,12 +1660,19 @@ ngx_stream_proxy_process(ngx_stream_session_t *s, ngx_uint_t from_upstream,
         if (size && src->read->ready && !src->read->delayed) {
 
             if (limit_rate) {
-                limit = (off_t) limit_rate * (ngx_time() - u->start_sec + 1)
-                        - *received;
+                ms = (ngx_msec_int_t) (ngx_current_msec - *limit_last);
+                ms = ngx_max(ms, 0);
+
+                excess = (off_t) (*limit_excess
+                                  - (uint64_t) limit_rate * ms / 1000);
+                excess = ngx_max(excess, 0);
+
+                limit = (off_t) limit_rate - excess;
 
                 if (limit <= 0) {
                     src->read->delayed = 1;
-                    delay = (ngx_msec_t) (- limit * 1000 / limit_rate + 1);
+                    excess -= (off_t) limit_rate / 2;
+                    delay = (ngx_msec_t) (excess * 1000 / limit_rate + 1);
                     ngx_add_timer(src->read, delay);
                     break;
                 }
@@ -1682,7 +1697,15 @@ ngx_stream_proxy_process(ngx_stream_session_t *s, ngx_uint_t from_upstream,
 
             if (n >= 0) {
                 if (limit_rate) {
-                    delay = (ngx_msec_t) (n * 1000 / limit_rate);
+                    excess += n;
+
+                    *limit_last = ngx_current_msec;
+                    *limit_excess = excess;
+
+                    excess -= (off_t) limit_rate / 2;
+                    excess = ngx_max(excess, 0);
+
+                    delay = (ngx_msec_t) (excess * 1000 / limit_rate);
 
                     if (delay > 0) {
                         src->read->delayed = 1;

src/stream/ngx_stream_upstream.h

@@ -127,14 +127,18 @@ typedef struct {
     ngx_chain_t                       *downstream_busy;
 
     off_t                              received;
-    time_t                             start_sec;
     ngx_uint_t                         requests;
     ngx_uint_t                         responses;
     ngx_msec_t                         start_time;
 
     size_t                             upload_rate;
     size_t                             download_rate;
 
+    ngx_msec_t                         upload_last;
+    ngx_msec_t                         download_last;
+    off_t                              upload_excess;
+    off_t                              download_excess;
+
     ngx_str_t                          ssl_name;
 
     ngx_stream_upstream_srv_conf_t    *upstream;