diff --git a/src/event/ngx_event.h b/src/event/ngx_event.h index 24c7435..5c1fd0d 100644 --- a/src/event/ngx_event.h +++ b/src/event/ngx_event.h @@ -507,6 +507,7 @@ void ngx_event_accept(ngx_event_t *ev); void ngx_event_recvmsg(ngx_event_t *ev); void ngx_udp_rbtree_insert_value(ngx_rbtree_node_t *temp, ngx_rbtree_node_t *node, ngx_rbtree_node_t *sentinel); +void ngx_delete_udp_connection(void *data); ngx_int_t ngx_trylock_accept_mutex(ngx_cycle_t *cycle); ngx_int_t ngx_enable_accept_events(ngx_cycle_t *cycle); u_char *ngx_accept_log_error(ngx_log_t *log, u_char *buf, size_t len); diff --git a/src/event/ngx_event_udp.c b/src/event/ngx_event_udp.c index 5bf131e..f4c7326 100644 --- a/src/event/ngx_event_udp.c +++ b/src/event/ngx_event_udp.c @@ -25,7 +25,6 @@ struct ngx_udp_connection_s { static ssize_t ngx_udp_shared_recv(ngx_connection_t *c, u_char *buf, size_t size); static ngx_int_t ngx_insert_udp_connection(ngx_connection_t *c); -static void ngx_delete_udp_connection(void *data); static ngx_connection_t *ngx_lookup_udp_connection(ngx_listening_t *ls, struct sockaddr *sockaddr, socklen_t socklen, struct sockaddr *local_sockaddr, socklen_t local_socklen); @@ -263,7 +262,10 @@ ngx_event_recvmsg(ngx_event_t *ev) rev->handler(rev); - c->udp->buffer = NULL; + if (c->udp) { + c->udp->buffer = NULL; + } + rev->ready = 0; goto next; @@ -540,12 +542,18 @@ ngx_insert_udp_connection(ngx_connection_t *c) } -static void +void ngx_delete_udp_connection(void *data) { ngx_connection_t *c = data; + if (c->udp == NULL) { + return; + } + ngx_rbtree_delete(&c->listening->rbtree, &c->udp->node); + + c->udp = NULL; } @@ -573,6 +581,14 @@ ngx_lookup_udp_connection(ngx_listening_t *ls, struct sockaddr *sockaddr, } } +#else + +void +ngx_delete_udp_connection(void *data) +{ + return; +} + #endif node = ls->rbtree.root; diff --git a/src/stream/ngx_stream_proxy_module.c b/src/stream/ngx_stream_proxy_module.c index 1282c9e..2a385cc 100644 --- a/src/stream/ngx_stream_proxy_module.c +++ b/src/stream/ngx_stream_proxy_module.c @@ -26,6 +26,7 @@ typedef struct { size_t buffer_size; size_t upload_rate; size_t download_rate; + ngx_uint_t requests; ngx_uint_t responses; ngx_uint_t next_upstream_tries; ngx_flag_t next_upstream; @@ -73,6 +74,8 @@ static void ngx_stream_proxy_connect_handler(ngx_event_t *ev); static ngx_int_t ngx_stream_proxy_test_connect(ngx_connection_t *c); static void ngx_stream_proxy_process(ngx_stream_session_t *s, ngx_uint_t from_upstream, ngx_uint_t do_write); +static ngx_int_t ngx_stream_proxy_test_finalize(ngx_stream_session_t *s, + ngx_uint_t from_upstream); static void ngx_stream_proxy_next_upstream(ngx_stream_session_t *s); static void ngx_stream_proxy_finalize(ngx_stream_session_t *s, ngx_uint_t rc); static u_char *ngx_stream_proxy_log_error(ngx_log_t *log, u_char *buf, @@ -193,6 +196,13 @@ static ngx_command_t ngx_stream_proxy_commands[] = { offsetof(ngx_stream_proxy_srv_conf_t, download_rate), NULL }, + { ngx_string("proxy_requests"), + NGX_STREAM_MAIN_CONF|NGX_STREAM_SRV_CONF|NGX_CONF_TAKE1, + ngx_conf_set_num_slot, + NGX_STREAM_SRV_CONF_OFFSET, + offsetof(ngx_stream_proxy_srv_conf_t, requests), + NULL }, + { ngx_string("proxy_responses"), NGX_STREAM_MAIN_CONF|NGX_STREAM_SRV_CONF|NGX_CONF_TAKE1, ngx_conf_set_num_slot, @@ -1339,11 +1349,14 @@ ngx_stream_proxy_process_connection(ngx_event_t *ev, ngx_uint_t from_upstream) } else { if (s->connection->type == SOCK_DGRAM) { - if (pscf->responses == NGX_MAX_INT32_VALUE) { + + if (pscf->responses == NGX_MAX_INT32_VALUE + || (u->responses >= pscf->responses * u->requests)) + { /* * successfully terminate timed out UDP session - * with unspecified number of responses + * if expected number of responses was received */ handler = c->log->handler; @@ -1646,44 +1659,7 @@ ngx_stream_proxy_process(ngx_stream_session_t *s, ngx_uint_t from_upstream, c->log->action = "proxying connection"; - if (c->type == SOCK_DGRAM - && pscf->responses != NGX_MAX_INT32_VALUE - && u->responses >= pscf->responses * u->requests - && !src->buffered && dst && !dst->buffered) - { - handler = c->log->handler; - c->log->handler = NULL; - - ngx_log_error(NGX_LOG_INFO, c->log, 0, - "udp done" - ", packets from/to client:%ui/%ui" - ", bytes from/to client:%O/%O" - ", bytes from/to upstream:%O/%O", - u->requests, u->responses, - s->received, c->sent, u->received, pc ? pc->sent : 0); - - c->log->handler = handler; - - ngx_stream_proxy_finalize(s, NGX_STREAM_OK); - return; - } - - if (c->type == SOCK_STREAM - && src->read->eof && dst && (dst->read->eof || !dst->buffered)) - { - handler = c->log->handler; - c->log->handler = NULL; - - ngx_log_error(NGX_LOG_INFO, c->log, 0, - "%s disconnected" - ", bytes from/to client:%O/%O" - ", bytes from/to upstream:%O/%O", - from_upstream ? "upstream" : "client", - s->received, c->sent, u->received, pc ? pc->sent : 0); - - c->log->handler = handler; - - ngx_stream_proxy_finalize(s, NGX_STREAM_OK); + if (ngx_stream_proxy_test_finalize(s, from_upstream) == NGX_OK) { return; } @@ -1710,6 +1686,87 @@ ngx_stream_proxy_process(ngx_stream_session_t *s, ngx_uint_t from_upstream, } +static ngx_int_t +ngx_stream_proxy_test_finalize(ngx_stream_session_t *s, + ngx_uint_t from_upstream) +{ + ngx_connection_t *c, *pc; + ngx_log_handler_pt handler; + ngx_stream_upstream_t *u; + ngx_stream_proxy_srv_conf_t *pscf; + + pscf = ngx_stream_get_module_srv_conf(s, ngx_stream_proxy_module); + + c = s->connection; + u = s->upstream; + pc = u->connected ? u->peer.connection : NULL; + + if (c->type == SOCK_DGRAM) { + + if (pscf->requests && u->requests < pscf->requests) { + return NGX_DECLINED; + } + + if (pscf->requests) { + ngx_delete_udp_connection(c); + } + + if (pscf->responses == NGX_MAX_INT32_VALUE + || u->responses < pscf->responses * u->requests) + { + return NGX_DECLINED; + } + + if (pc == NULL || c->buffered || pc->buffered) { + return NGX_DECLINED; + } + + handler = c->log->handler; + c->log->handler = NULL; + + ngx_log_error(NGX_LOG_INFO, c->log, 0, + "udp done" + ", packets from/to client:%ui/%ui" + ", bytes from/to client:%O/%O" + ", bytes from/to upstream:%O/%O", + u->requests, u->responses, + s->received, c->sent, u->received, pc ? pc->sent : 0); + + c->log->handler = handler; + + ngx_stream_proxy_finalize(s, NGX_STREAM_OK); + + return NGX_OK; + } + + /* c->type == SOCK_STREAM */ + + if (pc == NULL + || (!c->read->eof && !pc->read->eof) + || (!c->read->eof && c->buffered) + || (!pc->read->eof && pc->buffered)) + { + return NGX_DECLINED; + } + + handler = c->log->handler; + c->log->handler = NULL; + + ngx_log_error(NGX_LOG_INFO, c->log, 0, + "%s disconnected" + ", bytes from/to client:%O/%O" + ", bytes from/to upstream:%O/%O", + from_upstream ? "upstream" : "client", + s->received, c->sent, u->received, pc ? pc->sent : 0); + + c->log->handler = handler; + + ngx_stream_proxy_finalize(s, NGX_STREAM_OK); + + return NGX_OK; +} + + static void ngx_stream_proxy_next_upstream(ngx_stream_session_t *s) { @@ -1905,6 +1962,7 @@ ngx_stream_proxy_create_srv_conf(ngx_conf_t *cf) conf->buffer_size = NGX_CONF_UNSET_SIZE; conf->upload_rate = NGX_CONF_UNSET_SIZE; conf->download_rate = NGX_CONF_UNSET_SIZE; + conf->requests = NGX_CONF_UNSET_UINT; conf->responses = NGX_CONF_UNSET_UINT; conf->next_upstream_tries = NGX_CONF_UNSET_UINT; conf->next_upstream = NGX_CONF_UNSET; @@ -1949,6 +2007,9 @@ ngx_stream_proxy_merge_srv_conf(ngx_conf_t *cf, void *parent, void *child) ngx_conf_merge_size_value(conf->download_rate, prev->download_rate, 0); + ngx_conf_merge_uint_value(conf->requests, + prev->requests, 0); + ngx_conf_merge_uint_value(conf->responses, prev->responses, NGX_MAX_INT32_VALUE);