Mercurial > hg > nginx

comparison src/event/ngx_event_pipe.c @ 5883:973ee2276300

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Upstream: proxy_limit_rate and friends. The directives limit the upstream read rate. For example, "proxy_limit_rate 42" limits proxy upstream read rate to 42 bytes per second.
author Roman Arutyunyan <arut@nginx.com>
date Tue, 28 Oct 2014 12:29:59 +0300
parents ec81934727a1
children b8926ba4d087
comparison
equal deleted inserted replaced
5882:ec81934727a1 5883:973ee2276300
64 64
65 if (ngx_handle_read_event(rev, flags) != NGX_OK) { 65 if (ngx_handle_read_event(rev, flags) != NGX_OK) {
66 return NGX_ABORT; 66 return NGX_ABORT;
67 } 67 }
68 68
69 if (rev->active && !rev->ready) { 69 if (!rev->delayed) {
70 ngx_add_timer(rev, p->read_timeout); 70 if (rev->active && !rev->ready) {
71 71 ngx_add_timer(rev, p->read_timeout);
72 } else if (rev->timer_set) { 72
73 ngx_del_timer(rev); 73 } else if (rev->timer_set) {
74 ngx_del_timer(rev);
75 }
74 } 76 }
75 } 77 }
76 78
77 if (p->downstream->fd != (ngx_socket_t) -1 79 if (p->downstream->fd != (ngx_socket_t) -1
78 && p->downstream->data == p->output_ctx) 80 && p->downstream->data == p->output_ctx)
97 99
98 100
99 static ngx_int_t 101 static ngx_int_t
100 ngx_event_pipe_read_upstream(ngx_event_pipe_t *p) 102 ngx_event_pipe_read_upstream(ngx_event_pipe_t *p)
101 { 103 {
104 off_t limit;
102 ssize_t n, size; 105 ssize_t n, size;
103 ngx_int_t rc; 106 ngx_int_t rc;
104 ngx_buf_t *b; 107 ngx_buf_t *b;
108 ngx_msec_t delay;
105 ngx_chain_t *chain, *cl, *ln; 109 ngx_chain_t *chain, *cl, *ln;
106 110
107 if (p->upstream_eof || p->upstream_error || p->upstream_done) { 111 if (p->upstream_eof || p->upstream_error || p->upstream_done) {
108 return NGX_OK; 112 return NGX_OK;
109 } 113 }
166 } 170 }
167 171
168 break; 172 break;
169 } 173 }
170 #endif 174 #endif
175
176 if (p->limit_rate) {
177 if (p->upstream->read->delayed) {
178 break;
179 }
180
181 limit = (off_t) p->limit_rate * (ngx_time() - p->start_sec + 1)
182 - p->read_length;
183
184 if (limit <= 0) {
185 p->upstream->read->delayed = 1;
186 delay = (ngx_msec_t) (- limit * 1000 / p->limit_rate + 1);
187 ngx_add_timer(p->upstream->read, delay);
188 break;
189 }
190
191 } else {
192 limit = 0;
193 }
171 194
172 if (p->free_raw_bufs) { 195 if (p->free_raw_bufs) {
173 196
174 /* use the free bufs if they exist */ 197 /* use the free bufs if they exist */
175 198
268 "no pipe bufs to read in"); 291 "no pipe bufs to read in");
269 292
270 break; 293 break;
271 } 294 }
272 295
273 n = p->upstream->recv_chain(p->upstream, chain, 0); 296 n = p->upstream->recv_chain(p->upstream, chain, limit);
274 297
275 ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0, 298 ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0,
276 "pipe recv chain: %z", n); 299 "pipe recv chain: %z", n);
277 300
278 if (p->free_raw_bufs) { 301 if (p->free_raw_bufs) {
298 if (n == 0) { 321 if (n == 0) {
299 p->upstream_eof = 1; 322 p->upstream_eof = 1;
300 break; 323 break;
301 } 324 }
302 } 325 }
326
327 delay = p->limit_rate ? (ngx_msec_t) n * 1000 / p->limit_rate : 0;
303 328
304 p->read_length += n; 329 p->read_length += n;
305 cl = chain; 330 cl = chain;
306 p->free_raw_bufs = NULL; 331 p->free_raw_bufs = NULL;
307 332
334 if (cl) { 359 if (cl) {
335 for (ln = cl; ln->next; ln = ln->next) { /* void */ } 360 for (ln = cl; ln->next; ln = ln->next) { /* void */ }
336 361
337 ln->next = p->free_raw_bufs; 362 ln->next = p->free_raw_bufs;
338 p->free_raw_bufs = cl; 363 p->free_raw_bufs = cl;
364 }
365
366 if (delay > 0) {
367 p->upstream->read->delayed = 1;
368 ngx_add_timer(p->upstream->read, delay);
369 break;
339 } 370 }
340 } 371 }
341 372
342 #if (NGX_DEBUG) 373 #if (NGX_DEBUG)
343 374