nginx: src/event/ngx_event

comparison src/event/ngx_event_udp.c @ 7286:d27aa9060c95

Stream: udp streams. Previously, only one client packet could be processed in a udp stream session even though multiple response packets were supported. Now multiple packets coming from the same client address and port are delivered to the same stream session. If it's required to maintain a single stream of data, nginx should be configured in a way that all packets from a client are delivered to the same worker. On Linux and DragonFly BSD the "reuseport" parameter should be specified for this. Other systems do not currently provide appropriate mechanisms. For these systems a single stream of udp packets is only guaranteed in single-worker configurations. The proxy_response directive now specifies how many packets are expected in response to a single client packet.

author	Roman Arutyunyan <arut@nginx.com>
date	Mon, 04 Jun 2018 19:50:00 +0300
parents	88a624c9b491
children	27559d4a5151

comparison

equal deleted inserted replaced

-:88a624c9b491
+:d27aa9060c95
 #include <ngx_event.h>
 #if !(NGX_WIN32)
+struct ngx_udp_connection_s {
+ngx_rbtree_node_t   node;
+ngx_connection_t   *connection;
+ngx_buf_t          *buffer;
+};
 static void ngx_close_accepted_udp_connection(ngx_connection_t *c);
+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);
 void
 ngx_event_recvmsg(ngx_event_t *ev)
 {
 ssize_t            n;
+ngx_buf_t          buf;
 ngx_log_t         *log;
 ngx_err_t          err;
 socklen_t          socklen, local_socklen;
 ngx_event_t       *rev, *wev;
 struct iovec       iov[1];
 }
 }
 #endif
+c = ngx_lookup_udp_connection(ls, sockaddr, socklen, local_sockaddr,
+local_socklen);
+if (c) {
+#if (NGX_DEBUG)
+if (c->log->log_level & NGX_LOG_DEBUG_EVENT) {
+ngx_log_handler_pt  handler;
+handler = c->log->handler;
+c->log->handler = NULL;
+ngx_log_debug2(NGX_LOG_DEBUG_EVENT, c->log, 0,
+"recvmsg: fd:%d n:%z", c->fd, n);
+c->log->handler = handler;
+}
+#endif
+ngx_memzero(&buf, sizeof(ngx_buf_t));
+buf.pos = buffer;
+buf.last = buffer + n;
+rev = c->read;
+c->udp->buffer = &buf;
+rev->ready = 1;
+rev->handler(rev);
+c->udp->buffer = NULL;
+rev->ready = 0;
+goto next;
+}
 #if (NGX_STAT_STUB)
 (void) ngx_atomic_fetch_add(ngx_stat_accepted, 1);
 #endif
 ngx_accept_disabled = ngx_cycle->connection_n / 8
 return;
 }
 *log = ls->log;
+c->recv = ngx_udp_shared_recv;
 c->send = ngx_udp_send;
 c->send_chain = ngx_udp_send_chain;
 c->log = log;
 c->pool->log = log;
 }
 }
 #endif
+if (ngx_insert_udp_connection(c) != NGX_OK) {
+ngx_close_accepted_udp_connection(c);
+return;
+}
 log->data = NULL;
 log->handler = NULL;
 ls->handler(c);
+next:
 if (ngx_event_flags & NGX_USE_KQUEUE_EVENT) {
 ev->available -= n;
 }
 #if (NGX_STAT_STUB)
 (void) ngx_atomic_fetch_add(ngx_stat_active, -1);
 #endif
 }
-#endif
+static ssize_t
+ngx_udp_shared_recv(ngx_connection_t *c, u_char *buf, size_t size)
+{
+ssize_t     n;
+ngx_buf_t  *b;
+if (c->udp == NULL || c->udp->buffer == NULL) {
+return NGX_AGAIN;
+}
+b = c->udp->buffer;
+n = ngx_min(b->last - b->pos, (ssize_t) size);
+ngx_memcpy(buf, b->pos, n);
+c->udp->buffer = NULL;
+c->read->ready = 0;
+return n;
+}
+void
+ngx_udp_rbtree_insert_value(ngx_rbtree_node_t *temp,
+ngx_rbtree_node_t *node, ngx_rbtree_node_t *sentinel)
+{
+ngx_int_t               rc;
+ngx_connection_t       *c, *ct;
+ngx_rbtree_node_t     **p;
+ngx_udp_connection_t   *udp, *udpt;
+for ( ;; ) {
+if (node->key < temp->key) {
+p = &temp->left;
+} else if (node->key > temp->key) {
+p = &temp->right;
+} else { /* node->key == temp->key */
+udp = (ngx_udp_connection_t *) node;
+c = udp->connection;
+udpt = (ngx_udp_connection_t *) temp;
+ct = udpt->connection;
+rc = ngx_cmp_sockaddr(c->sockaddr, c->socklen,
+ct->sockaddr, ct->socklen, 1);
+if (rc == 0 && c->listening->wildcard) {
+rc = ngx_cmp_sockaddr(c->local_sockaddr, c->local_socklen,
+ct->local_sockaddr, ct->local_socklen, 1);
+}
+p = (rc < 0) ? &temp->left : &temp->right;
+}
+if (*p == sentinel) {
+break;
+}
+temp = *p;
+}
+*p = node;
+node->parent = temp;
+node->left = sentinel;
+node->right = sentinel;
+ngx_rbt_red(node);
+}
+static ngx_int_t
+ngx_insert_udp_connection(ngx_connection_t *c)
+{
+uint32_t               hash;
+ngx_pool_cleanup_t    *cln;
+ngx_udp_connection_t  *udp;
+if (c->udp) {
+return NGX_OK;
+}
+udp = ngx_pcalloc(c->pool, sizeof(ngx_udp_connection_t));
+if (udp == NULL) {
+return NGX_ERROR;
+}
+udp->connection = c;
+ngx_crc32_init(hash);
+ngx_crc32_update(&hash, (u_char *) c->sockaddr, c->socklen);
+if (c->listening->wildcard) {
+ngx_crc32_update(&hash, (u_char *) c->local_sockaddr, c->local_socklen);
+}
+ngx_crc32_final(hash);
+udp->node.key = hash;
+cln = ngx_pool_cleanup_add(c->pool, 0);
+if (cln == NULL) {
+return NGX_ERROR;
+}
+cln->data = c;
+cln->handler = ngx_delete_udp_connection;
+ngx_rbtree_insert(&c->listening->rbtree, &udp->node);
+c->udp = udp;
+return NGX_OK;
+}
+static void
+ngx_delete_udp_connection(void *data)
+{
+ngx_connection_t  *c = data;
+ngx_rbtree_delete(&c->listening->rbtree, &c->udp->node);
+}
+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)
+{
+uint32_t               hash;
+ngx_int_t              rc;
+ngx_connection_t      *c;
+ngx_rbtree_node_t     *node, *sentinel;
+ngx_udp_connection_t  *udp;
+#if (NGX_HAVE_UNIX_DOMAIN)
+if (sockaddr->sa_family == AF_UNIX) {
+struct sockaddr_un *saun = (struct sockaddr_un *) sockaddr;
+if (socklen <= (socklen_t) offsetof(struct sockaddr_un, sun_path)
+|| saun->sun_path[0] == '\0')
+{
+ngx_log_debug0(NGX_LOG_DEBUG_EVENT, ngx_cycle->log, 0,
+"unbound unix socket");
+return NULL;
+}
+}
+#endif
+node = ls->rbtree.root;
+sentinel = ls->rbtree.sentinel;
+ngx_crc32_init(hash);
+ngx_crc32_update(&hash, (u_char *) sockaddr, socklen);
+if (ls->wildcard) {
+ngx_crc32_update(&hash, (u_char *) local_sockaddr, local_socklen);
+}
+ngx_crc32_final(hash);
+while (node != sentinel) {
+if (hash < node->key) {
+node = node->left;
+continue;
+}
+if (hash > node->key) {
+node = node->right;
+continue;
+}
+/* hash == node->key */
+udp = (ngx_udp_connection_t *) node;
+c = udp->connection;
+rc = ngx_cmp_sockaddr(sockaddr, socklen,
+c->sockaddr, c->socklen, 1);
+if (rc == 0 && ls->wildcard) {
+rc = ngx_cmp_sockaddr(local_sockaddr, local_socklen,
+c->local_sockaddr, c->local_socklen, 1);
+}
+if (rc == 0) {
+return c;
+}
+node = (rc < 0) ? node->left : node->right;
+}
+return NULL;
+}
+#endif

Mercurial > hg > nginx

comparison src/event/ngx_event_udp.c @ 7286:d27aa9060c95