Ticket #1298: udp_session_reuse.diff

nginx-1.12.0/src/core/ngx_buf.c

@@ -43 +43 @@
     return b;
 }
 
+ngx_buf_t *
+ngx_recreate_temp_buf(ngx_buf_t *b, ngx_pool_t *pool, size_t size)
+{
+    if (b == NULL) {
+        return NULL;
+    }
+
+    if (b->start != NULL && ngx_pfree(pool, b->start) != NGX_OK) {
+        return NULL;
+    }
+
+    ngx_memzero(b, sizeof(ngx_buf_t));
+    b->start = ngx_palloc(pool, size);
+    if (b->start == NULL) {
+        return NULL;
+    }
+
+    /*
+     * set by ngx_calloc_buf():
+     *
+     *     b->file_pos = 0;
+     *     b->file_last = 0;
+     *     b->file = NULL;
+     *     b->shadow = NULL;
+     *     b->tag = 0;
+     *     and flags
+     */
+
+    b->pos = b->start;
+    b->last = b->start;
+    b->end = b->last + size;
+    b->temporary = 1;
+
+    return b;
+}
+
 
 ngx_chain_t *
 ngx_alloc_chain_link(ngx_pool_t *pool)

nginx-1.12.0/src/core/ngx_buf.h

@@ -141 +141 @@
                             (b->file_last - b->file_pos))
 
 ngx_buf_t *ngx_create_temp_buf(ngx_pool_t *pool, size_t size);
+ngx_buf_t *ngx_recreate_temp_buf(ngx_buf_t *b, ngx_pool_t *pool, size_t size);
 ngx_chain_t *ngx_create_chain_of_bufs(ngx_pool_t *pool, ngx_bufs_t *bufs);
 
 

nginx-1.12.0/src/core/ngx_connection.c

@@ -1158 +1158 @@
 
     log_error = c->log_error;
 
+    if (c->type == SOCK_DGRAM && c->listening != NULL) {
+        uint64_t client_key = sock2key((struct sockaddr_in *)c->sockaddr);
+        ngx_radix64tree_delete(c->listening->dgram_conns, client_key, client_key);
+    }
+
     ngx_free_connection(c);
 
     fd = c->fd;

nginx-1.12.0/src/core/ngx_connection.h

@@ -87 +87 @@
     int                 fastopen;
 #endif
 
+    ngx_radix_tree_t   *dgram_conns;
 };
 
 

nginx-1.12.0/src/core/ngx_radix_tree.c

@@ -262 +262 @@
     return value;
 }
 
+ngx_int_t
+ngx_radix64tree_insert(ngx_radix_tree_t *tree, uint64_t key, uint64_t mask,
+    uintptr_t value)
+{
+    uint64_t           bit;
+    ngx_radix_node_t  *node, *next;
+
+    bit = 0x8000000000000000;
+
+    node = tree->root;
+    next = tree->root;
+
+    while (bit & mask) {
+        if (key & bit) {
+            next = node->right;
+
+        } else {
+            next = node->left;
+        }
+
+        if (next == NULL) {
+            break;
+        }
+
+        bit >>= 1;
+        node = next;
+    }
+
+    if (next) {
+        if (node->value != NGX_RADIX_NO_VALUE) {
+            return NGX_BUSY;
+        }
+
+        node->value = value;
+        return NGX_OK;
+    }
+
+    while (bit & mask) {
+        next = ngx_radix_alloc(tree);
+        if (next == NULL) {
+            return NGX_ERROR;
+        }
+
+        next->right = NULL;
+        next->left = NULL;
+        next->parent = node;
+        next->value = NGX_RADIX_NO_VALUE;
+
+        if (key & bit) {
+            node->right = next;
+
+        } else {
+            node->left = next;
+        }
+
+        bit >>= 1;
+        node = next;
+    }
+
+    node->value = value;
+
+    return NGX_OK;
+}
+
+
+ngx_int_t
+ngx_radix64tree_delete(ngx_radix_tree_t *tree, uint64_t key, uint64_t mask)
+{
+    uint64_t           bit;
+    ngx_radix_node_t  *node;
+
+    bit = 0x8000000000000000;
+    node = tree->root;
+
+    while (node && (bit & mask)) {
+        if (key & bit) {
+            node = node->right;
+
+        } else {
+            node = node->left;
+        }
+
+        bit >>= 1;
+    }
+
+    if (node == NULL) {
+        return NGX_ERROR;
+    }
+
+    if (node->right || node->left) {
+        if (node->value != NGX_RADIX_NO_VALUE) {
+            node->value = NGX_RADIX_NO_VALUE;
+            return NGX_OK;
+        }
+
+        return NGX_ERROR;
+    }
+
+    for ( ;; ) {
+        if (node->parent->right == node) {
+            node->parent->right = NULL;
+
+        } else {
+            node->parent->left = NULL;
+        }
+
+        node->right = tree->free;
+        tree->free = node;
+
+        node = node->parent;
+
+        if (node->right || node->left) {
+            break;
+        }
+
+        if (node->value != NGX_RADIX_NO_VALUE) {
+            break;
+        }
+
+        if (node->parent == NULL) {
+            break;
+        }
+    }
+
+    return NGX_OK;
+}
+
+
+uintptr_t
+ngx_radix64tree_find(ngx_radix_tree_t *tree, uint64_t key)
+{
+    uint64_t           bit;
+    uintptr_t          value;
+    ngx_radix_node_t  *node;
+
+    bit = 0x8000000000000000;
+    value = NGX_RADIX_NO_VALUE;
+    node = tree->root;
+
+    while (node) {
+        if (node->value != NGX_RADIX_NO_VALUE) {
+            value = node->value;
+        }
+
+        if (key & bit) {
+            node = node->right;
+
+        } else {
+            node = node->left;
+        }
+
+        bit >>= 1;
+    }
+
+    return value;
+}
+
 
 #if (NGX_HAVE_INET6)
 

nginx-1.12.0/src/core/ngx_radix_tree.h

@@ -43 +43 @@
     uint32_t key, uint32_t mask);
 uintptr_t ngx_radix32tree_find(ngx_radix_tree_t *tree, uint32_t key);
 
+ngx_int_t ngx_radix64tree_insert(ngx_radix_tree_t *tree,
+    uint64_t key, uint64_t mask, uintptr_t value);
+ngx_int_t ngx_radix64tree_delete(ngx_radix_tree_t *tree,
+    uint64_t key, uint64_t mask);
+uintptr_t ngx_radix64tree_find(ngx_radix_tree_t *tree, uint64_t key);
+
 #if (NGX_HAVE_INET6)
 ngx_int_t ngx_radix128tree_insert(ngx_radix_tree_t *tree,
     u_char *key, u_char *mask, uintptr_t value);
@@ -51 +57 @@
 uintptr_t ngx_radix128tree_find(ngx_radix_tree_t *tree, u_char *key);
 #endif
 
+static ngx_inline uint64_t
+sock2key(struct sockaddr_in *sin)
+{
+    return ((uint64_t)sin->sin_addr.s_addr << 32) + sin->sin_port;
+}
 
 #endif /* _NGX_RADIX_TREE_H_INCLUDED_ */

nginx-1.12.0/src/event/ngx_event_accept.c

@@ -21 +21 @@
     ngx_connection_t *c);
 #endif
 
+void ngx_stream_session_udp_handler(ngx_event_t *rev);
+
 
 void
 ngx_event_accept(ngx_event_t *ev)
@@ -341 +343 @@
     ngx_event_conf_t  *ecf;
     ngx_connection_t  *c, *lc;
     static u_char      buffer[65535];
+#if (NGX_DEBUG)
+    ngx_str_t          client_addr;
+    u_char             client_text[NGX_SOCKADDR_STRLEN];
+#endif
+    uint64_t           client_key;
+    uintptr_t          v;
 
 #if (NGX_HAVE_MSGHDR_MSG_CONTROL)
 
@@ -440 +448 @@
         ngx_accept_disabled = ngx_cycle->connection_n / 8
                               - ngx_cycle->free_connection_n;
 
+#if (NGX_DEBUG)
+        client_addr.data = client_text;
+        client_addr.len = ngx_sock_ntop((struct sockaddr *)msg.msg_name,
+                msg.msg_namelen, client_text, NGX_SOCKADDR_STRLEN, 1);
+#endif
+
+        client_key = sock2key(&sa.sockaddr_in);
+        v = ngx_radix64tree_find(ls->dgram_conns, client_key);
+        if (v != NGX_RADIX_NO_VALUE) {
+#if (NGX_DEBUG)
+            ngx_log_debug3(NGX_LOG_DEBUG_EVENT, ev->log, 0,
+                "reuse incoming conenction on %V from %V(%08X)", &ls->addr_text, &client_addr, client_key);
+#endif
+            c = (ngx_connection_t *)v;
+            c->number = ngx_atomic_fetch_add(ngx_connection_counter, 1);
+            c->log->connection = c->number;
+            c->buffer = ngx_recreate_temp_buf(c->buffer, c->pool, n);
+            if (c->buffer == NULL) {
+                ngx_log_debug1(NGX_LOG_DEBUG_EVENT, ev->log, 0,
+                    "failed to recreate buf for incoming conenction on %V", &ls->addr_text);
+                ngx_close_accepted_connection(c);
+                return;
+            }
+
+            c->buffer->last = ngx_cpymem(c->buffer->last, buffer, n);
+            ngx_stream_session_udp_handler(c->read);
+            continue;
+        }
+
+#if (NGX_DEBUG)
+        ngx_log_debug3(NGX_LOG_DEBUG_EVENT, ev->log, 0,
+            "setup incoming conenction on %V from %V(%08X)", &ls->addr_text, &client_addr, client_key);
+#endif
         c = ngx_get_connection(lc->fd, ev->log);
         if (c == NULL) {
 #ifdef NGX_HTTP_STATUS
@@ -641 +682 @@
             ev->available -= n;
         }
 
+        client_key = sock2key(&sa.sockaddr_in);
+        ngx_radix64tree_insert(ls->dgram_conns, client_key, client_key, (uintptr_t)c);
     } while (ev->available);
 }
 

nginx-1.12.0/src/stream/ngx_stream.c

@@ -521 +521 @@
 
             stport->naddrs = i + 1;
 
+            if (ls->type == SOCK_DGRAM) {
+                ls->dgram_conns = ngx_radix_tree_create(cf->pool, -1);
+                if (ls->dgram_conns == NULL) {
+                    return NGX_CONF_ERROR;
+                }
+            } else {
+                ls->dgram_conns = NULL;
+            }
+
             switch (ls->sockaddr->sa_family) {
 #if (NGX_HAVE_INET6)
             case AF_INET6:

nginx-1.12.0/src/stream/ngx_stream.h

@@ -293 +293 @@
 
 void ngx_stream_init_connection(ngx_connection_t *c);
 void ngx_stream_session_handler(ngx_event_t *rev);
+void ngx_stream_session_udp_handler(ngx_event_t *rev);
 void ngx_stream_finalize_session(ngx_stream_session_t *s, ngx_uint_t rc);
 
 

nginx-1.12.0/src/stream/ngx_stream_handler.c

@@ -336 +336 @@
     ngx_stream_core_run_phases(s);
 }
 
+void
+ngx_stream_session_udp_handler(ngx_event_t *rev)
+{
+    ngx_connection_t      *c;
+    ngx_stream_session_t  *s;
+
+    c = rev->data;
+    s = c->data;
+
+    ngx_log_debug0(NGX_LOG_DEBUG_STREAM, c->log, 0,
+                   "stream session udp handler");
+
+    c->log->action = "reuse session";
+
+    s->received += c->buffer->last - c->buffer->pos;
+
+    ngx_stream_core_run_phases(s);
+}
+
 
 void
 ngx_stream_finalize_session(ngx_stream_session_t *s, ngx_uint_t rc)

nginx-1.12.0/src/stream/ngx_stream_proxy_module.c

@@ -368 +368 @@
 
     c = s->connection;
 
+    if (c->type == SOCK_DGRAM && s->upstream) {
+        ngx_stream_proxy_connect(s);
+        return;
+    }
+
     pscf = ngx_stream_get_module_srv_conf(s, ngx_stream_proxy_module);
 
     ngx_log_debug0(NGX_LOG_DEBUG_STREAM, c->log, 0,
@@ -709 +714 @@
 
     c = s->connection;
 
-    c->log->action = "connecting to upstream";
-
     pscf = ngx_stream_get_module_srv_conf(s, ngx_stream_proxy_module);
 
     u = s->upstream;
 
+    if (c->type == SOCK_DGRAM && u->peer.connection) {
+        ngx_chain_t              *cl;
+
+        pc = u->peer.connection;
+        
+        ngx_log_debug1(NGX_LOG_DEBUG_STREAM, c->log, 0, "reuse proxy connect: %p", pc);
+
+        if (c->buffer && c->buffer->pos < c->buffer->last) {
+            ngx_log_debug1(NGX_LOG_DEBUG_STREAM, c->log, 0,
+                           "stream proxy add buffer: %uz",
+                           c->buffer->last - c->buffer->pos);
+
+            cl = ngx_chain_get_free_buf(c->pool, &u->free);
+            if (cl == NULL) {
+#ifdef NGX_HTTP_STATUS
+                ngx_http_status_update(cscf->status_index, STATUS_TYPE_PROXY_ERROR, 1);
+#endif
+                ngx_stream_proxy_finalize(s, NGX_STREAM_INTERNAL_SERVER_ERROR);
+                return;
+            }
+
+            *cl->buf = *c->buffer;
+
+            cl->buf->tag = (ngx_buf_tag_t) &ngx_stream_proxy_module;
+            cl->buf->flush = 1;
+            cl->buf->last_buf = 1;
+
+            cl->next = u->upstream_out;
+            u->upstream_out = cl;
+        }
+
+        if (pscf->responses == 0) {
+            pc->read->ready = 0;
+            pc->read->eof = 1;
+        }
+
+        ngx_stream_proxy_process(s, 0, 1);
+    
+        return;
+    }
+
+    c->log->action = "connecting to upstream";
+
     u->connected = 0;
     u->proxy_protocol = pscf->proxy_protocol;