# HG changeset patch # User Roman Arutyunyan # Date 1527861349 -10800 # Node ID 88a624c9b4917d43a381716885c6415069d1e504 # Parent 52aacc8ddcc54879442c707dab0625cfc4b3a44f Events: moved ngx_recvmsg() to new file src/event/ngx_event_udp.c. diff --git a/auto/sources b/auto/sources --- a/auto/sources +++ b/auto/sources @@ -95,6 +95,7 @@ EVENT_SRCS="src/event/ngx_event.c \ src/event/ngx_event_timer.c \ src/event/ngx_event_posted.c \ src/event/ngx_event_accept.c \ + src/event/ngx_event_udp.c \ src/event/ngx_event_connect.c \ src/event/ngx_event_pipe.c" diff --git a/src/event/ngx_event.h b/src/event/ngx_event.h --- a/src/event/ngx_event.h +++ b/src/event/ngx_event.h @@ -508,7 +508,11 @@ void ngx_event_accept(ngx_event_t *ev); void ngx_event_recvmsg(ngx_event_t *ev); #endif 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); +#if (NGX_DEBUG) +void ngx_debug_accepted_connection(ngx_event_conf_t *ecf, ngx_connection_t *c); +#endif void ngx_process_events_and_timers(ngx_cycle_t *cycle); diff --git a/src/event/ngx_event_accept.c b/src/event/ngx_event_accept.c --- a/src/event/ngx_event_accept.c +++ b/src/event/ngx_event_accept.c @@ -10,13 +10,8 @@ #include -static ngx_int_t ngx_enable_accept_events(ngx_cycle_t *cycle); static ngx_int_t ngx_disable_accept_events(ngx_cycle_t *cycle, ngx_uint_t all); static void ngx_close_accepted_connection(ngx_connection_t *c); -#if (NGX_DEBUG) -static void ngx_debug_accepted_connection(ngx_event_conf_t *ecf, - ngx_connection_t *c); -#endif void @@ -320,352 +315,6 @@ ngx_event_accept(ngx_event_t *ev) } -#if !(NGX_WIN32) - -void -ngx_event_recvmsg(ngx_event_t *ev) -{ - ssize_t n; - ngx_log_t *log; - ngx_err_t err; - socklen_t socklen, local_socklen; - ngx_event_t *rev, *wev; - struct iovec iov[1]; - struct msghdr msg; - ngx_sockaddr_t sa, lsa; - struct sockaddr *sockaddr, *local_sockaddr; - ngx_listening_t *ls; - ngx_event_conf_t *ecf; - ngx_connection_t *c, *lc; - static u_char buffer[65535]; - -#if (NGX_HAVE_MSGHDR_MSG_CONTROL) - -#if (NGX_HAVE_IP_RECVDSTADDR) - u_char msg_control[CMSG_SPACE(sizeof(struct in_addr))]; -#elif (NGX_HAVE_IP_PKTINFO) - u_char msg_control[CMSG_SPACE(sizeof(struct in_pktinfo))]; -#endif - -#if (NGX_HAVE_INET6 && NGX_HAVE_IPV6_RECVPKTINFO) - u_char msg_control6[CMSG_SPACE(sizeof(struct in6_pktinfo))]; -#endif - -#endif - - if (ev->timedout) { - if (ngx_enable_accept_events((ngx_cycle_t *) ngx_cycle) != NGX_OK) { - return; - } - - ev->timedout = 0; - } - - ecf = ngx_event_get_conf(ngx_cycle->conf_ctx, ngx_event_core_module); - - if (!(ngx_event_flags & NGX_USE_KQUEUE_EVENT)) { - ev->available = ecf->multi_accept; - } - - lc = ev->data; - ls = lc->listening; - ev->ready = 0; - - ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0, - "recvmsg on %V, ready: %d", &ls->addr_text, ev->available); - - do { - ngx_memzero(&msg, sizeof(struct msghdr)); - - iov[0].iov_base = (void *) buffer; - iov[0].iov_len = sizeof(buffer); - - msg.msg_name = &sa; - msg.msg_namelen = sizeof(ngx_sockaddr_t); - msg.msg_iov = iov; - msg.msg_iovlen = 1; - -#if (NGX_HAVE_MSGHDR_MSG_CONTROL) - - if (ls->wildcard) { - -#if (NGX_HAVE_IP_RECVDSTADDR || NGX_HAVE_IP_PKTINFO) - if (ls->sockaddr->sa_family == AF_INET) { - msg.msg_control = &msg_control; - msg.msg_controllen = sizeof(msg_control); - } -#endif - -#if (NGX_HAVE_INET6 && NGX_HAVE_IPV6_RECVPKTINFO) - if (ls->sockaddr->sa_family == AF_INET6) { - msg.msg_control = &msg_control6; - msg.msg_controllen = sizeof(msg_control6); - } -#endif - } - -#endif - - n = recvmsg(lc->fd, &msg, 0); - - if (n == -1) { - err = ngx_socket_errno; - - if (err == NGX_EAGAIN) { - ngx_log_debug0(NGX_LOG_DEBUG_EVENT, ev->log, err, - "recvmsg() not ready"); - return; - } - - ngx_log_error(NGX_LOG_ALERT, ev->log, err, "recvmsg() failed"); - - return; - } - -#if (NGX_HAVE_MSGHDR_MSG_CONTROL) - if (msg.msg_flags & (MSG_TRUNC|MSG_CTRUNC)) { - ngx_log_error(NGX_LOG_ALERT, ev->log, 0, - "recvmsg() truncated data"); - continue; - } -#endif - - sockaddr = msg.msg_name; - socklen = msg.msg_namelen; - - if (socklen > (socklen_t) sizeof(ngx_sockaddr_t)) { - socklen = sizeof(ngx_sockaddr_t); - } - - if (socklen == 0) { - - /* - * on Linux recvmsg() returns zero msg_namelen - * when receiving packets from unbound AF_UNIX sockets - */ - - socklen = sizeof(struct sockaddr); - ngx_memzero(&sa, sizeof(struct sockaddr)); - sa.sockaddr.sa_family = ls->sockaddr->sa_family; - } - - local_sockaddr = ls->sockaddr; - local_socklen = ls->socklen; - -#if (NGX_HAVE_MSGHDR_MSG_CONTROL) - - if (ls->wildcard) { - struct cmsghdr *cmsg; - - ngx_memcpy(&lsa, local_sockaddr, local_socklen); - local_sockaddr = &lsa.sockaddr; - - for (cmsg = CMSG_FIRSTHDR(&msg); - cmsg != NULL; - cmsg = CMSG_NXTHDR(&msg, cmsg)) - { - -#if (NGX_HAVE_IP_RECVDSTADDR) - - if (cmsg->cmsg_level == IPPROTO_IP - && cmsg->cmsg_type == IP_RECVDSTADDR - && local_sockaddr->sa_family == AF_INET) - { - struct in_addr *addr; - struct sockaddr_in *sin; - - addr = (struct in_addr *) CMSG_DATA(cmsg); - sin = (struct sockaddr_in *) local_sockaddr; - sin->sin_addr = *addr; - - break; - } - -#elif (NGX_HAVE_IP_PKTINFO) - - if (cmsg->cmsg_level == IPPROTO_IP - && cmsg->cmsg_type == IP_PKTINFO - && local_sockaddr->sa_family == AF_INET) - { - struct in_pktinfo *pkt; - struct sockaddr_in *sin; - - pkt = (struct in_pktinfo *) CMSG_DATA(cmsg); - sin = (struct sockaddr_in *) local_sockaddr; - sin->sin_addr = pkt->ipi_addr; - - break; - } - -#endif - -#if (NGX_HAVE_INET6 && NGX_HAVE_IPV6_RECVPKTINFO) - - if (cmsg->cmsg_level == IPPROTO_IPV6 - && cmsg->cmsg_type == IPV6_PKTINFO - && local_sockaddr->sa_family == AF_INET6) - { - struct in6_pktinfo *pkt6; - struct sockaddr_in6 *sin6; - - pkt6 = (struct in6_pktinfo *) CMSG_DATA(cmsg); - sin6 = (struct sockaddr_in6 *) local_sockaddr; - sin6->sin6_addr = pkt6->ipi6_addr; - - break; - } - -#endif - - } - } - -#endif - -#if (NGX_STAT_STUB) - (void) ngx_atomic_fetch_add(ngx_stat_accepted, 1); -#endif - - ngx_accept_disabled = ngx_cycle->connection_n / 8 - - ngx_cycle->free_connection_n; - - c = ngx_get_connection(lc->fd, ev->log); - if (c == NULL) { - return; - } - - c->shared = 1; - c->type = SOCK_DGRAM; - c->socklen = socklen; - -#if (NGX_STAT_STUB) - (void) ngx_atomic_fetch_add(ngx_stat_active, 1); -#endif - - c->pool = ngx_create_pool(ls->pool_size, ev->log); - if (c->pool == NULL) { - ngx_close_accepted_connection(c); - return; - } - - c->sockaddr = ngx_palloc(c->pool, socklen); - if (c->sockaddr == NULL) { - ngx_close_accepted_connection(c); - return; - } - - ngx_memcpy(c->sockaddr, sockaddr, socklen); - - log = ngx_palloc(c->pool, sizeof(ngx_log_t)); - if (log == NULL) { - ngx_close_accepted_connection(c); - return; - } - - *log = ls->log; - - c->send = ngx_udp_send; - c->send_chain = ngx_udp_send_chain; - - c->log = log; - c->pool->log = log; - c->listening = ls; - - if (local_sockaddr == &lsa.sockaddr) { - local_sockaddr = ngx_palloc(c->pool, local_socklen); - if (local_sockaddr == NULL) { - ngx_close_accepted_connection(c); - return; - } - - ngx_memcpy(local_sockaddr, &lsa, local_socklen); - } - - c->local_sockaddr = local_sockaddr; - c->local_socklen = local_socklen; - - c->buffer = ngx_create_temp_buf(c->pool, n); - if (c->buffer == NULL) { - ngx_close_accepted_connection(c); - return; - } - - c->buffer->last = ngx_cpymem(c->buffer->last, buffer, n); - - rev = c->read; - wev = c->write; - - wev->ready = 1; - - rev->log = log; - wev->log = log; - - /* - * TODO: MT: - ngx_atomic_fetch_add() - * or protection by critical section or light mutex - * - * TODO: MP: - allocated in a shared memory - * - ngx_atomic_fetch_add() - * or protection by critical section or light mutex - */ - - c->number = ngx_atomic_fetch_add(ngx_connection_counter, 1); - -#if (NGX_STAT_STUB) - (void) ngx_atomic_fetch_add(ngx_stat_handled, 1); -#endif - - if (ls->addr_ntop) { - c->addr_text.data = ngx_pnalloc(c->pool, ls->addr_text_max_len); - if (c->addr_text.data == NULL) { - ngx_close_accepted_connection(c); - return; - } - - c->addr_text.len = ngx_sock_ntop(c->sockaddr, c->socklen, - c->addr_text.data, - ls->addr_text_max_len, 0); - if (c->addr_text.len == 0) { - ngx_close_accepted_connection(c); - return; - } - } - -#if (NGX_DEBUG) - { - ngx_str_t addr; - u_char text[NGX_SOCKADDR_STRLEN]; - - ngx_debug_accepted_connection(ecf, c); - - if (log->log_level & NGX_LOG_DEBUG_EVENT) { - addr.data = text; - addr.len = ngx_sock_ntop(c->sockaddr, c->socklen, text, - NGX_SOCKADDR_STRLEN, 1); - - ngx_log_debug4(NGX_LOG_DEBUG_EVENT, log, 0, - "*%uA recvmsg: %V fd:%d n:%z", - c->number, &addr, c->fd, n); - } - - } -#endif - - log->data = NULL; - log->handler = NULL; - - ls->handler(c); - - if (ngx_event_flags & NGX_USE_KQUEUE_EVENT) { - ev->available -= n; - } - - } while (ev->available); -} - -#endif - - ngx_int_t ngx_trylock_accept_mutex(ngx_cycle_t *cycle) { @@ -704,7 +353,7 @@ ngx_trylock_accept_mutex(ngx_cycle_t *cy } -static ngx_int_t +ngx_int_t ngx_enable_accept_events(ngx_cycle_t *cycle) { ngx_uint_t i; @@ -779,7 +428,7 @@ ngx_close_accepted_connection(ngx_connec fd = c->fd; c->fd = (ngx_socket_t) -1; - if (!c->shared && ngx_close_socket(fd) == -1) { + if (ngx_close_socket(fd) == -1) { ngx_log_error(NGX_LOG_ALERT, c->log, ngx_socket_errno, ngx_close_socket_n " failed"); } @@ -804,7 +453,7 @@ ngx_accept_log_error(ngx_log_t *log, u_c #if (NGX_DEBUG) -static void +void ngx_debug_accepted_connection(ngx_event_conf_t *ecf, ngx_connection_t *c) { struct sockaddr_in *sin; diff --git a/src/event/ngx_event_udp.c b/src/event/ngx_event_udp.c new file mode 100644 --- /dev/null +++ b/src/event/ngx_event_udp.c @@ -0,0 +1,376 @@ + +/* + * Copyright (C) Roman Arutyunyan + * Copyright (C) Nginx, Inc. + */ + + +#include +#include +#include + + +#if !(NGX_WIN32) + +static void ngx_close_accepted_udp_connection(ngx_connection_t *c); + + +void +ngx_event_recvmsg(ngx_event_t *ev) +{ + ssize_t n; + ngx_log_t *log; + ngx_err_t err; + socklen_t socklen, local_socklen; + ngx_event_t *rev, *wev; + struct iovec iov[1]; + struct msghdr msg; + ngx_sockaddr_t sa, lsa; + struct sockaddr *sockaddr, *local_sockaddr; + ngx_listening_t *ls; + ngx_event_conf_t *ecf; + ngx_connection_t *c, *lc; + static u_char buffer[65535]; + +#if (NGX_HAVE_MSGHDR_MSG_CONTROL) + +#if (NGX_HAVE_IP_RECVDSTADDR) + u_char msg_control[CMSG_SPACE(sizeof(struct in_addr))]; +#elif (NGX_HAVE_IP_PKTINFO) + u_char msg_control[CMSG_SPACE(sizeof(struct in_pktinfo))]; +#endif + +#if (NGX_HAVE_INET6 && NGX_HAVE_IPV6_RECVPKTINFO) + u_char msg_control6[CMSG_SPACE(sizeof(struct in6_pktinfo))]; +#endif + +#endif + + if (ev->timedout) { + if (ngx_enable_accept_events((ngx_cycle_t *) ngx_cycle) != NGX_OK) { + return; + } + + ev->timedout = 0; + } + + ecf = ngx_event_get_conf(ngx_cycle->conf_ctx, ngx_event_core_module); + + if (!(ngx_event_flags & NGX_USE_KQUEUE_EVENT)) { + ev->available = ecf->multi_accept; + } + + lc = ev->data; + ls = lc->listening; + ev->ready = 0; + + ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0, + "recvmsg on %V, ready: %d", &ls->addr_text, ev->available); + + do { + ngx_memzero(&msg, sizeof(struct msghdr)); + + iov[0].iov_base = (void *) buffer; + iov[0].iov_len = sizeof(buffer); + + msg.msg_name = &sa; + msg.msg_namelen = sizeof(ngx_sockaddr_t); + msg.msg_iov = iov; + msg.msg_iovlen = 1; + +#if (NGX_HAVE_MSGHDR_MSG_CONTROL) + + if (ls->wildcard) { + +#if (NGX_HAVE_IP_RECVDSTADDR || NGX_HAVE_IP_PKTINFO) + if (ls->sockaddr->sa_family == AF_INET) { + msg.msg_control = &msg_control; + msg.msg_controllen = sizeof(msg_control); + } +#endif + +#if (NGX_HAVE_INET6 && NGX_HAVE_IPV6_RECVPKTINFO) + if (ls->sockaddr->sa_family == AF_INET6) { + msg.msg_control = &msg_control6; + msg.msg_controllen = sizeof(msg_control6); + } +#endif + } + +#endif + + n = recvmsg(lc->fd, &msg, 0); + + if (n == -1) { + err = ngx_socket_errno; + + if (err == NGX_EAGAIN) { + ngx_log_debug0(NGX_LOG_DEBUG_EVENT, ev->log, err, + "recvmsg() not ready"); + return; + } + + ngx_log_error(NGX_LOG_ALERT, ev->log, err, "recvmsg() failed"); + + return; + } + +#if (NGX_HAVE_MSGHDR_MSG_CONTROL) + if (msg.msg_flags & (MSG_TRUNC|MSG_CTRUNC)) { + ngx_log_error(NGX_LOG_ALERT, ev->log, 0, + "recvmsg() truncated data"); + continue; + } +#endif + + sockaddr = msg.msg_name; + socklen = msg.msg_namelen; + + if (socklen > (socklen_t) sizeof(ngx_sockaddr_t)) { + socklen = sizeof(ngx_sockaddr_t); + } + + if (socklen == 0) { + + /* + * on Linux recvmsg() returns zero msg_namelen + * when receiving packets from unbound AF_UNIX sockets + */ + + socklen = sizeof(struct sockaddr); + ngx_memzero(&sa, sizeof(struct sockaddr)); + sa.sockaddr.sa_family = ls->sockaddr->sa_family; + } + + local_sockaddr = ls->sockaddr; + local_socklen = ls->socklen; + +#if (NGX_HAVE_MSGHDR_MSG_CONTROL) + + if (ls->wildcard) { + struct cmsghdr *cmsg; + + ngx_memcpy(&lsa, local_sockaddr, local_socklen); + local_sockaddr = &lsa.sockaddr; + + for (cmsg = CMSG_FIRSTHDR(&msg); + cmsg != NULL; + cmsg = CMSG_NXTHDR(&msg, cmsg)) + { + +#if (NGX_HAVE_IP_RECVDSTADDR) + + if (cmsg->cmsg_level == IPPROTO_IP + && cmsg->cmsg_type == IP_RECVDSTADDR + && local_sockaddr->sa_family == AF_INET) + { + struct in_addr *addr; + struct sockaddr_in *sin; + + addr = (struct in_addr *) CMSG_DATA(cmsg); + sin = (struct sockaddr_in *) local_sockaddr; + sin->sin_addr = *addr; + + break; + } + +#elif (NGX_HAVE_IP_PKTINFO) + + if (cmsg->cmsg_level == IPPROTO_IP + && cmsg->cmsg_type == IP_PKTINFO + && local_sockaddr->sa_family == AF_INET) + { + struct in_pktinfo *pkt; + struct sockaddr_in *sin; + + pkt = (struct in_pktinfo *) CMSG_DATA(cmsg); + sin = (struct sockaddr_in *) local_sockaddr; + sin->sin_addr = pkt->ipi_addr; + + break; + } + +#endif + +#if (NGX_HAVE_INET6 && NGX_HAVE_IPV6_RECVPKTINFO) + + if (cmsg->cmsg_level == IPPROTO_IPV6 + && cmsg->cmsg_type == IPV6_PKTINFO + && local_sockaddr->sa_family == AF_INET6) + { + struct in6_pktinfo *pkt6; + struct sockaddr_in6 *sin6; + + pkt6 = (struct in6_pktinfo *) CMSG_DATA(cmsg); + sin6 = (struct sockaddr_in6 *) local_sockaddr; + sin6->sin6_addr = pkt6->ipi6_addr; + + break; + } + +#endif + + } + } + +#endif + +#if (NGX_STAT_STUB) + (void) ngx_atomic_fetch_add(ngx_stat_accepted, 1); +#endif + + ngx_accept_disabled = ngx_cycle->connection_n / 8 + - ngx_cycle->free_connection_n; + + c = ngx_get_connection(lc->fd, ev->log); + if (c == NULL) { + return; + } + + c->shared = 1; + c->type = SOCK_DGRAM; + c->socklen = socklen; + +#if (NGX_STAT_STUB) + (void) ngx_atomic_fetch_add(ngx_stat_active, 1); +#endif + + c->pool = ngx_create_pool(ls->pool_size, ev->log); + if (c->pool == NULL) { + ngx_close_accepted_udp_connection(c); + return; + } + + c->sockaddr = ngx_palloc(c->pool, socklen); + if (c->sockaddr == NULL) { + ngx_close_accepted_udp_connection(c); + return; + } + + ngx_memcpy(c->sockaddr, sockaddr, socklen); + + log = ngx_palloc(c->pool, sizeof(ngx_log_t)); + if (log == NULL) { + ngx_close_accepted_udp_connection(c); + return; + } + + *log = ls->log; + + c->send = ngx_udp_send; + c->send_chain = ngx_udp_send_chain; + + c->log = log; + c->pool->log = log; + c->listening = ls; + + if (local_sockaddr == &lsa.sockaddr) { + local_sockaddr = ngx_palloc(c->pool, local_socklen); + if (local_sockaddr == NULL) { + ngx_close_accepted_udp_connection(c); + return; + } + + ngx_memcpy(local_sockaddr, &lsa, local_socklen); + } + + c->local_sockaddr = local_sockaddr; + c->local_socklen = local_socklen; + + c->buffer = ngx_create_temp_buf(c->pool, n); + if (c->buffer == NULL) { + ngx_close_accepted_udp_connection(c); + return; + } + + c->buffer->last = ngx_cpymem(c->buffer->last, buffer, n); + + rev = c->read; + wev = c->write; + + wev->ready = 1; + + rev->log = log; + wev->log = log; + + /* + * TODO: MT: - ngx_atomic_fetch_add() + * or protection by critical section or light mutex + * + * TODO: MP: - allocated in a shared memory + * - ngx_atomic_fetch_add() + * or protection by critical section or light mutex + */ + + c->number = ngx_atomic_fetch_add(ngx_connection_counter, 1); + +#if (NGX_STAT_STUB) + (void) ngx_atomic_fetch_add(ngx_stat_handled, 1); +#endif + + if (ls->addr_ntop) { + c->addr_text.data = ngx_pnalloc(c->pool, ls->addr_text_max_len); + if (c->addr_text.data == NULL) { + ngx_close_accepted_udp_connection(c); + return; + } + + c->addr_text.len = ngx_sock_ntop(c->sockaddr, c->socklen, + c->addr_text.data, + ls->addr_text_max_len, 0); + if (c->addr_text.len == 0) { + ngx_close_accepted_udp_connection(c); + return; + } + } + +#if (NGX_DEBUG) + { + ngx_str_t addr; + u_char text[NGX_SOCKADDR_STRLEN]; + + ngx_debug_accepted_connection(ecf, c); + + if (log->log_level & NGX_LOG_DEBUG_EVENT) { + addr.data = text; + addr.len = ngx_sock_ntop(c->sockaddr, c->socklen, text, + NGX_SOCKADDR_STRLEN, 1); + + ngx_log_debug4(NGX_LOG_DEBUG_EVENT, log, 0, + "*%uA recvmsg: %V fd:%d n:%z", + c->number, &addr, c->fd, n); + } + + } +#endif + + log->data = NULL; + log->handler = NULL; + + ls->handler(c); + + if (ngx_event_flags & NGX_USE_KQUEUE_EVENT) { + ev->available -= n; + } + + } while (ev->available); +} + + +static void +ngx_close_accepted_udp_connection(ngx_connection_t *c) +{ + ngx_free_connection(c); + + c->fd = (ngx_socket_t) -1; + + if (c->pool) { + ngx_destroy_pool(c->pool); + } + +#if (NGX_STAT_STUB) + (void) ngx_atomic_fetch_add(ngx_stat_active, -1); +#endif +} + +#endif