One Level Up Top Level

src/event/ngx_event_udp.c - nginx source code

Functions defined

Source code


  2. /*
  3. * Copyright (C) Roman Arutyunyan
  4. * Copyright (C) Nginx, Inc.
  5. */


  8. #include <ngx_config.h>
  9. #include <ngx_core.h>
  10. #include <ngx_event.h>


  13. #if !(NGX_WIN32)

  15. static void ngx_close_accepted_udp_connection(ngx_connection_t *c);
  16. static ssize_t ngx_udp_shared_recv(ngx_connection_t *c, u_char *buf,
  17.     size_t size);
  18. static ngx_int_t ngx_insert_udp_connection(ngx_connection_t *c);
  19. static ngx_connection_t *ngx_lookup_udp_connection(ngx_listening_t *ls,
  20.     struct sockaddr *sockaddr, socklen_t socklen,
  21.     struct sockaddr *local_sockaddr, socklen_t local_socklen);


  24. void
  25. ‌ngx_event_recvmsg(ngx_event_t *ev)
  26. {
  27.     ssize_t            n;
  28.     ngx_buf_t          buf;
  29.     ngx_log_t         *log;
  30.     ngx_err_t          err;
  31.     socklen_t          socklen, local_socklen;
  32.     ngx_event_t       *rev, *wev;
  33.     struct iovec       iov[1];
  34.     struct msghdr      msg;
  35.     ngx_sockaddr_t     sa, lsa;
  36.     struct sockaddr   *sockaddr, *local_sockaddr;
  37.     ngx_listening_t   *ls;
  38.     ngx_event_conf_t  *ecf;
  39.     ngx_connection_t  *c, *lc;
  40.     static u_char      buffer[65535];

  42. #if (NGX_HAVE_ADDRINFO_CMSG)
  43.     u_char             msg_control[CMSG_SPACE(sizeof(ngx_addrinfo_t))];
  44. #endif

  46.     if (ev->timedout) {
  47.         if (ngx_enable_accept_events((ngx_cycle_t *) ngx_cycle) != NGX_OK) {
  48.             return;
  49.         }

  51.         ev->timedout = 0;
  52.     }

  54.     ecf = ngx_event_get_conf(ngx_cycle->conf_ctx, ngx_event_core_module);

  56.     if (!(ngx_event_flags & NGX_USE_KQUEUE_EVENT)) {
  57.         ev->available = ecf->multi_accept;
  58.     }

  60.     lc = ev->data;
  61.     ls = lc->listening;
  62.     ev->ready = 0;

  64.     ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
  65.                    "recvmsg on %V, ready: %d", &ls->addr_text, ev->available);

  67.     do {
  68.         ngx_memzero(&msg, sizeof(struct msghdr));

  70.         iov[0].iov_base = (void *) buffer;
  71.         iov[0].iov_len = sizeof(buffer);

  73.         msg.msg_name = &sa;
  74.         msg.msg_namelen = sizeof(ngx_sockaddr_t);
  75.         msg.msg_iov = iov;
  76.         msg.msg_iovlen = 1;

  78. #if (NGX_HAVE_ADDRINFO_CMSG)
  79.         if (ls->wildcard) {
  80.             msg.msg_control = &msg_control;
  81.             msg.msg_controllen = sizeof(msg_control);

  83.             ngx_memzero(&msg_control, sizeof(msg_control));
  84.         }
  85. #endif

  87.         n = recvmsg(lc->fd, &msg, 0);

  89.         if (n == -1) {
  90.             err = ngx_socket_errno;

  92.             if (err == NGX_EAGAIN) {
  93.                 ngx_log_debug0(NGX_LOG_DEBUG_EVENT, ev->log, err,
  94.                                "recvmsg() not ready");
  95.                 return;
  96.             }

  98.             ngx_log_error(NGX_LOG_ALERT, ev->log, err, "recvmsg() failed");

  100.             return;
  101.         }

  103. #if (NGX_HAVE_ADDRINFO_CMSG)
  104.         if (msg.msg_flags & (MSG_TRUNC|MSG_CTRUNC)) {
  105.             ngx_log_error(NGX_LOG_ALERT, ev->log, 0,
  106.                           "recvmsg() truncated data");
  107.             continue;
  108.         }
  109. #endif

  111.         sockaddr = msg.msg_name;
  112.         socklen = msg.msg_namelen;

  114.         if (socklen > (socklen_t) sizeof(ngx_sockaddr_t)) {
  115.             socklen = sizeof(ngx_sockaddr_t);
  116.         }

  118.         if (socklen == 0) {

  120.             /*
  121.              * on Linux recvmsg() returns zero msg_namelen
  122.              * when receiving packets from unbound AF_UNIX sockets
  123.              */

  125.             socklen = sizeof(struct sockaddr);
  126.             ngx_memzero(&sa, sizeof(struct sockaddr));
  127.             sa.sockaddr.sa_family = ls->sockaddr->sa_family;
  128.         }

  130.         local_sockaddr = ls->sockaddr;
  131.         local_socklen = ls->socklen;

  133. #if (NGX_HAVE_ADDRINFO_CMSG)

  135.         if (ls->wildcard) {
  136.             struct cmsghdr  *cmsg;

  138.             ngx_memcpy(&lsa, local_sockaddr, local_socklen);
  139.             local_sockaddr = &lsa.sockaddr;

  141.             for (cmsg = CMSG_FIRSTHDR(&msg);
  142.                  cmsg != NULL;
  143.                  cmsg = CMSG_NXTHDR(&msg, cmsg))
  144.             {
  145.                 if (ngx_get_srcaddr_cmsg(cmsg, local_sockaddr) == NGX_OK) {
  146.                     break;
  147.                 }
  148.             }
  149.         }

  151. #endif

  153.         c = ngx_lookup_udp_connection(ls, sockaddr, socklen, local_sockaddr,
  154.                                       local_socklen);

  156.         if (c) {

  158. #if (NGX_DEBUG)
  159.             if (c->log->log_level & NGX_LOG_DEBUG_EVENT) {
  160.                 ngx_log_handler_pt  handler;

  162.                 handler = c->log->handler;
  163.                 c->log->handler = NULL;

  165.                 ngx_log_debug2(NGX_LOG_DEBUG_EVENT, c->log, 0,
  166.                                "recvmsg: fd:%d n:%z", c->fd, n);

  168.                 c->log->handler = handler;
  169.             }
  170. #endif

  172.             ngx_memzero(&buf, sizeof(ngx_buf_t));

  174.             buf.pos = buffer;
  175.             buf.last = buffer + n;

  177.             rev = c->read;

  179.             c->udp->buffer = &buf;

  181.             rev->ready = 1;
  182.             rev->active = 0;

  184.             rev->handler(rev);

  186.             if (c->udp) {
  187.                 c->udp->buffer = NULL;
  188.             }

  190.             rev->ready = 0;
  191.             rev->active = 1;

  193.             goto next;
  194.         }

  196. #if (NGX_STAT_STUB)
  197.         (void) ngx_atomic_fetch_add(ngx_stat_accepted, 1);
  198. #endif

  200.         ngx_accept_disabled = ngx_cycle->connection_n / 8
  201.                               - ngx_cycle->free_connection_n;

  203.         c = ngx_get_connection(lc->fd, ev->log);
  204.         if (c == NULL) {
  205.             return;
  206.         }

  208.         c->shared = 1;
  209.         c->type = SOCK_DGRAM;
  210.         c->socklen = socklen;

  212. #if (NGX_STAT_STUB)
  213.         (void) ngx_atomic_fetch_add(ngx_stat_active, 1);
  214. #endif

  216.         c->pool = ngx_create_pool(ls->pool_size, ev->log);
  217.         if (c->pool == NULL) {
  218.             ngx_close_accepted_udp_connection(c);
  219.             return;
  220.         }

  222.         c->sockaddr = ngx_palloc(c->pool, socklen);
  223.         if (c->sockaddr == NULL) {
  224.             ngx_close_accepted_udp_connection(c);
  225.             return;
  226.         }

  228.         ngx_memcpy(c->sockaddr, sockaddr, socklen);

  230.         log = ngx_palloc(c->pool, sizeof(ngx_log_t));
  231.         if (log == NULL) {
  232.             ngx_close_accepted_udp_connection(c);
  233.             return;
  234.         }

  236.         *log = ls->log;

  238.         c->recv = ngx_udp_shared_recv;
  239.         c->send = ngx_udp_send;
  240.         c->send_chain = ngx_udp_send_chain;

  242.         c->need_flush_buf = 1;

  244.         c->log = log;
  245.         c->pool->log = log;
  246.         c->listening = ls;

  248.         if (local_sockaddr == &lsa.sockaddr) {
  249.             local_sockaddr = ngx_palloc(c->pool, local_socklen);
  250.             if (local_sockaddr == NULL) {
  251.                 ngx_close_accepted_udp_connection(c);
  252.                 return;
  253.             }

  255.             ngx_memcpy(local_sockaddr, &lsa, local_socklen);
  256.         }

  258.         c->local_sockaddr = local_sockaddr;
  259.         c->local_socklen = local_socklen;

  261.         c->buffer = ngx_create_temp_buf(c->pool, n);
  262.         if (c->buffer == NULL) {
  263.             ngx_close_accepted_udp_connection(c);
  264.             return;
  265.         }

  267.         c->buffer->last = ngx_cpymem(c->buffer->last, buffer, n);

  269.         rev = c->read;
  270.         wev = c->write;

  272.         rev->active = 1;
  273.         wev->ready = 1;

  275.         rev->log = log;
  276.         wev->log = log;

  278.         /*
  279.          * TODO: MT: - ngx_atomic_fetch_add()
  280.          *             or protection by critical section or light mutex
  281.          *
  282.          * TODO: MP: - allocated in a shared memory
  283.          *           - ngx_atomic_fetch_add()
  284.          *             or protection by critical section or light mutex
  285.          */

  287.         c->number = ngx_atomic_fetch_add(ngx_connection_counter, 1);

  289.         c->start_time = ngx_current_msec;

  291. #if (NGX_STAT_STUB)
  292.         (void) ngx_atomic_fetch_add(ngx_stat_handled, 1);
  293. #endif

  295.         if (ls->addr_ntop) {
  296.             c->addr_text.data = ngx_pnalloc(c->pool, ls->addr_text_max_len);
  297.             if (c->addr_text.data == NULL) {
  298.                 ngx_close_accepted_udp_connection(c);
  299.                 return;
  300.             }

  302.             c->addr_text.len = ngx_sock_ntop(c->sockaddr, c->socklen,
  303.                                              c->addr_text.data,
  304.                                              ls->addr_text_max_len, 0);
  305.             if (c->addr_text.len == 0) {
  306.                 ngx_close_accepted_udp_connection(c);
  307.                 return;
  308.             }
  309.         }

  311. #if (NGX_DEBUG)
  312.         {
  313.         ngx_str_t  addr;
  314.         u_char     text[NGX_SOCKADDR_STRLEN];

  316.         ngx_debug_accepted_connection(ecf, c);

  318.         if (log->log_level & NGX_LOG_DEBUG_EVENT) {
  319.             addr.data = text;
  320.             addr.len = ngx_sock_ntop(c->sockaddr, c->socklen, text,
  321.                                      NGX_SOCKADDR_STRLEN, 1);

  323.             ngx_log_debug4(NGX_LOG_DEBUG_EVENT, log, 0,
  324.                            "*%uA recvmsg: %V fd:%d n:%z",
  325.                            c->number, &addr, c->fd, n);
  326.         }

  328.         }
  329. #endif

  331.         if (ngx_insert_udp_connection(c) != NGX_OK) {
  332.             ngx_close_accepted_udp_connection(c);
  333.             return;
  334.         }

  336.         log->data = NULL;
  337.         log->handler = NULL;

  339.         ls->handler(c);

  341.     next:

  343.         if (ngx_event_flags & NGX_USE_KQUEUE_EVENT) {
  344.             ev->available -= n;
  345.         }

  347.     } while (ev->available);
  348. }


  351. static void
  352. ‌ngx_close_accepted_udp_connection(ngx_connection_t *c)
  353. {
  354.     ngx_free_connection(c);

  356.     c->fd = (ngx_socket_t) -1;

  358.     if (c->pool) {
  359.         ngx_destroy_pool(c->pool);
  360.     }

  362. #if (NGX_STAT_STUB)
  363.     (void) ngx_atomic_fetch_add(ngx_stat_active, -1);
  364. #endif
  365. }


  368. static ssize_t
  369. ‌ngx_udp_shared_recv(ngx_connection_t *c, u_char *buf, size_t size)
  370. {
  371.     ssize_t     n;
  372.     ngx_buf_t  *b;

  374.     if (c->udp == NULL || c->udp->buffer == NULL) {
  375.         return NGX_AGAIN;
  376.     }

  378.     b = c->udp->buffer;

  380.     n = ngx_min(b->last - b->pos, (ssize_t) size);

  382.     ngx_memcpy(buf, b->pos, n);

  384.     c->udp->buffer = NULL;

  386.     c->read->ready = 0;
  387.     c->read->active = 1;

  389.     return n;
  390. }


  393. void
  394. ‌ngx_udp_rbtree_insert_value(ngx_rbtree_node_t *temp,
  395.     ngx_rbtree_node_t *node, ngx_rbtree_node_t *sentinel)
  396. {
  397.     ngx_int_t               rc;
  398.     ngx_connection_t       *c, *ct;
  399.     ngx_rbtree_node_t     **p;
  400.     ngx_udp_connection_t   *udp, *udpt;

  402.     for ( ;; ) {

  404.         if (node->key < temp->key) {

  406.             p = &temp->left;

  408.         } else if (node->key > temp->key) {

  410.             p = &temp->right;

  412.         } else { /* node->key == temp->key */

  414.             udp = (ngx_udp_connection_t *) node;
  415.             c = udp->connection;

  417.             udpt = (ngx_udp_connection_t *) temp;
  418.             ct = udpt->connection;

  420.             rc = ngx_memn2cmp(udp->key.data, udpt->key.data,
  421.                               udp->key.len, udpt->key.len);

  423.             if (rc == 0 && c->listening->wildcard) {
  424.                 rc = ngx_cmp_sockaddr(c->local_sockaddr, c->local_socklen,
  425.                                       ct->local_sockaddr, ct->local_socklen, 1);
  426.             }

  428.             p = (rc < 0) ? &temp->left : &temp->right;
  429.         }

  431.         if (*p == sentinel) {
  432.             break;
  433.         }

  435.         temp = *p;
  436.     }

  438.     *p = node;
  439.     node->parent = temp;
  440.     node->left = sentinel;
  441.     node->right = sentinel;
  442.     ngx_rbt_red(node);
  443. }


  446. static ngx_int_t
  447. ‌ngx_insert_udp_connection(ngx_connection_t *c)
  448. {
  449.     uint32_t               hash;
  450.     ngx_pool_cleanup_t    *cln;
  451.     ngx_udp_connection_t  *udp;

  453.     if (c->udp) {
  454.         return NGX_OK;
  455.     }

  457.     udp = ngx_pcalloc(c->pool, sizeof(ngx_udp_connection_t));
  458.     if (udp == NULL) {
  459.         return NGX_ERROR;
  460.     }

  462.     udp->connection = c;

  464.     ngx_crc32_init(hash);
  465.     ngx_crc32_update(&hash, (u_char *) c->sockaddr, c->socklen);

  467.     if (c->listening->wildcard) {
  468.         ngx_crc32_update(&hash, (u_char *) c->local_sockaddr, c->local_socklen);
  469.     }

  471.     ngx_crc32_final(hash);

  473.     udp->node.key = hash;
  474.     udp->key.data = (u_char *) c->sockaddr;
  475.     udp->key.len = c->socklen;

  477.     cln = ngx_pool_cleanup_add(c->pool, 0);
  478.     if (cln == NULL) {
  479.         return NGX_ERROR;
  480.     }

  482.     cln->data = c;
  483.     cln->handler = ngx_delete_udp_connection;

  485.     ngx_rbtree_insert(&c->listening->rbtree, &udp->node);

  487.     c->udp = udp;

  489.     return NGX_OK;
  490. }


  493. void
  494. ‌ngx_delete_udp_connection(void *data)
  495. {
  496.     ngx_connection_t  *c = data;

  498.     if (c->udp == NULL) {
  499.         return;
  500.     }

  502.     ngx_rbtree_delete(&c->listening->rbtree, &c->udp->node);

  504.     c->udp = NULL;
  505. }


  508. static ngx_connection_t *
  509. ‌ngx_lookup_udp_connection(ngx_listening_t *ls, struct sockaddr *sockaddr,
  510.     socklen_t socklen, struct sockaddr *local_sockaddr, socklen_t local_socklen)
  511. {
  512.     uint32_t               hash;
  513.     ngx_int_t              rc;
  514.     ngx_connection_t      *c;
  515.     ngx_rbtree_node_t     *node, *sentinel;
  516.     ngx_udp_connection_t  *udp;

  518. #if (NGX_HAVE_UNIX_DOMAIN)

  520.     if (sockaddr->sa_family == AF_UNIX) {
  521.         struct sockaddr_un *saun = (struct sockaddr_un *) sockaddr;

  523.         if (socklen <= (socklen_t) offsetof(struct sockaddr_un, sun_path)
  524.             || saun->sun_path[0] == '\0')
  525.         {
  526.             ngx_log_debug0(NGX_LOG_DEBUG_EVENT, ngx_cycle->log, 0,
  527.                            "unbound unix socket");
  528.             return NULL;
  529.         }
  530.     }

  532. #endif

  534.     node = ls->rbtree.root;
  535.     sentinel = ls->rbtree.sentinel;

  537.     ngx_crc32_init(hash);
  538.     ngx_crc32_update(&hash, (u_char *) sockaddr, socklen);

  540.     if (ls->wildcard) {
  541.         ngx_crc32_update(&hash, (u_char *) local_sockaddr, local_socklen);
  542.     }

  544.     ngx_crc32_final(hash);

  546.     while (node != sentinel) {

  548.         if (hash < node->key) {
  549.             node = node->left;
  550.             continue;
  551.         }

  553.         if (hash > node->key) {
  554.             node = node->right;
  555.             continue;
  556.         }

  558.         /* hash == node->key */

  560.         udp = (ngx_udp_connection_t *) node;

  562.         c = udp->connection;

  564.         rc = ngx_cmp_sockaddr(sockaddr, socklen,
  565.                               c->sockaddr, c->socklen, 1);

  567.         if (rc == 0 && ls->wildcard) {
  568.             rc = ngx_cmp_sockaddr(local_sockaddr, local_socklen,
  569.                                   c->local_sockaddr, c->local_socklen, 1);
  570.         }

  572.         if (rc == 0) {
  573.             return c;
  574.         }

  576.         node = (rc < 0) ? node->left : node->right;
  577.     }

  579.     return NULL;
  580. }

  582. #else

  584. void
  585. ‌ngx_delete_udp_connection(void *data)
  586. {
  587.     return;
  588. }

  590. #endif

One Level Up Top Level