ngx_udp_recv()

auto/sources

@@ -164,6 +164,7 @@ UNIX_SRCS="$CORE_SRCS $EVENT_SRCS \
             src/os/unix/ngx_socket.c \
             src/os/unix/ngx_recv.c \
             src/os/unix/ngx_readv_chain.c \
+            src/os/unix/ngx_udp_recv.c \
             src/os/unix/ngx_send.c \
             src/os/unix/ngx_writev_chain.c \
             src/os/unix/ngx_channel.c \
@@ -229,6 +230,7 @@ WIN32_SRCS="$CORE_SRCS $EVENT_SRCS \
             src/os/win32/ngx_socket.c \
             src/os/win32/ngx_wsarecv.c \
             src/os/win32/ngx_wsarecv_chain.c \
+            src/os/win32/ngx_udp_wsarecv.c \
             src/os/win32/ngx_wsasend_chain.c \
             src/os/win32/ngx_win32_init.c \
             src/os/win32/ngx_user.c \

src/core/ngx_resolver.c

@@ -826,9 +826,9 @@ ngx_resolver_read_response(ngx_event_t *rev)
     c = rev->data;
 
     do {
-        n = ngx_recv(c, buf, NGX_RESOLVER_UDP_SIZE);
+        n = ngx_udp_recv(c, buf, NGX_RESOLVER_UDP_SIZE);
 
-        if (n == -1) {
+        if (n < 0) {
             return;
         }
 

src/event/modules/ngx_aio_module.c

@@ -28,6 +28,7 @@ static ngx_int_t ngx_aio_process_events(ngx_cycle_t *cycle, ngx_msec_t timer,
 ngx_os_io_t ngx_os_aio = {
     ngx_aio_read,
     ngx_aio_read_chain,
+    NULL,
     ngx_aio_write,
     ngx_aio_write_chain,
     0

src/event/modules/ngx_iocp_module.c

@@ -90,6 +90,7 @@ ngx_module_t  ngx_iocp_module = {
 ngx_os_io_t ngx_iocp_io = {
     ngx_overlapped_wsarecv,
     NULL,
+    ngx_udp_overlapped_wsarecv,
     NULL,
     ngx_overlapped_wsasend_chain,
     0

src/event/ngx_event.h

@@ -429,6 +429,7 @@ extern ngx_os_io_t  ngx_io;
 
 #define ngx_recv             ngx_io.recv
 #define ngx_recv_chain       ngx_io.recv_chain
+#define ngx_udp_recv         ngx_io.udp_recv
 #define ngx_send             ngx_io.send
 #define ngx_send_chain       ngx_io.send_chain
 

src/os/unix/ngx_freebsd_init.c

@@ -30,6 +30,7 @@ ngx_uint_t ngx_freebsd_use_tcp_nopush;
 static ngx_os_io_t ngx_freebsd_io = {
     ngx_unix_recv,
     ngx_readv_chain,
+    ngx_udp_unix_recv,
     ngx_unix_send,
 #if (NGX_HAVE_SENDFILE)
     ngx_freebsd_sendfile_chain,

src/os/unix/ngx_linux_init.c

@@ -17,6 +17,7 @@ int   ngx_linux_rtsig_max;
 static ngx_os_io_t ngx_linux_io = {
     ngx_unix_recv,
     ngx_readv_chain,
+    ngx_udp_unix_recv,
     ngx_unix_send,
 #if (NGX_HAVE_SENDFILE)
     ngx_linux_sendfile_chain,

src/os/unix/ngx_os.h

@@ -25,6 +25,7 @@ typedef ngx_chain_t *(*ngx_send_chain_pt)(ngx_connection_t *c, ngx_chain_t *in,
 typedef struct {
     ngx_recv_pt        recv;
     ngx_recv_chain_pt  recv_chain;
+    ngx_recv_pt        udp_recv;
     ngx_send_pt        send;
     ngx_send_chain_pt  send_chain;
     ngx_uint_t         flags;
@@ -41,6 +42,7 @@ ngx_int_t ngx_daemon(ngx_log_t *log);
 
 ssize_t ngx_unix_recv(ngx_connection_t *c, u_char *buf, size_t size);
 ssize_t ngx_readv_chain(ngx_connection_t *c, ngx_chain_t *entry);
+ssize_t ngx_udp_unix_recv(ngx_connection_t *c, u_char *buf, size_t size);
 ssize_t ngx_unix_send(ngx_connection_t *c, u_char *buf, size_t size);
 ngx_chain_t *ngx_writev_chain(ngx_connection_t *c, ngx_chain_t *in,
     off_t limit);

src/os/unix/ngx_posix_init.c

@@ -21,6 +21,7 @@ struct rlimit  rlmt;
 ngx_os_io_t ngx_os_io = {
     ngx_unix_recv,
     ngx_readv_chain,
+    ngx_udp_unix_recv,
     NULL,
     ngx_writev_chain,
     0

src/os/unix/ngx_solaris_init.c

@@ -16,6 +16,7 @@ char ngx_solaris_version[50];
 static ngx_os_io_t ngx_solaris_io = {
     ngx_unix_recv,
     ngx_readv_chain,
+    ngx_udp_unix_recv,
     ngx_unix_send,
 #if (NGX_HAVE_SENDFILE)
     ngx_solaris_sendfilev_chain,

src/os/unix/ngx_udp_recv.c

@@ -0,0 +1,114 @@
+
+/*
+ * Copyright (C) Igor Sysoev
+ */
+
+
+#include <ngx_config.h>
+#include <ngx_core.h>
+#include <ngx_event.h>
+
+
+#if (NGX_HAVE_KQUEUE)
+
+ssize_t
+ngx_udp_unix_recv(ngx_connection_t *c, u_char *buf, size_t size)
+{
+    ssize_t       n;
+    ngx_err_t     err;
+    ngx_event_t  *rev;
+
+    rev = c->read;
+
+    do {
+        n = recv(c->fd, buf, size, 0);
+
+        ngx_log_debug3(NGX_LOG_DEBUG_EVENT, c->log, 0,
+                       "recv: fd:%d %d of %d", c->fd, n, size);
+
+        if (n >= 0) {
+            if (ngx_event_flags & NGX_USE_KQUEUE_EVENT) {
+                rev->available -= n;
+
+                /*
+                 * rev->available may be negative here because some additional
+                 * bytes may be received between kevent() and recv()
+                 */
+
+                if (rev->available <= 0) {
+                    rev->ready = 0;
+                    rev->available = 0;
+                }
+            }
+
+            return n;
+        }
+
+        err = ngx_socket_errno;
+
+        if (err == NGX_EAGAIN || err == NGX_EINTR) {
+            ngx_log_debug0(NGX_LOG_DEBUG_EVENT, c->log, err,
+                           "recv() not ready");
+            n = NGX_AGAIN;
+
+        } else {
+            n = ngx_connection_error(c, err, "recv() failed");
+            break;
+        }
+
+    } while (err == NGX_EINTR);
+
+    rev->ready = 0;
+
+    if (n == NGX_ERROR){
+        rev->error = 1;
+    }
+
+    return n;
+}
+
+#else /* ! NGX_HAVE_KQUEUE */
+
+ssize_t
+ngx_udp_unix_recv(ngx_connection_t *c, u_char *buf, size_t size)
+{
+    ssize_t       n;
+    ngx_err_t     err;
+    ngx_event_t  *rev;
+
+    rev = c->read;
+
+    do {
+        n = recv(c->fd, buf, size, 0);
+
+        ngx_log_debug3(NGX_LOG_DEBUG_EVENT, c->log, 0,
+                       "recv: fd:%d %d of %d", c->fd, n, size);
+
+        if (n >= 0) {
+            return n;
+        }
+
+        err = ngx_socket_errno;
+
+        if (err == NGX_EAGAIN || err == NGX_EINTR) {
+            ngx_log_debug0(NGX_LOG_DEBUG_EVENT, c->log, err,
+                           "recv() not ready");
+            n = NGX_AGAIN;
+
+        } else {
+            n = ngx_connection_error(c, err, "recv() failed");
+            break;
+        }
+
+    } while (err == NGX_EINTR);
+
+    rev->ready = 0;
+
+    if (n == NGX_ERROR){
+        rev->error = 1;
+    }
+
+    return n;
+}
+
+#endif /* NGX_HAVE_KQUEUE */

src/os/win32/ngx_os.h

@@ -20,11 +20,12 @@ typedef ssize_t (*ngx_recv_pt)(ngx_connection_t *c, u_char *buf, size_t size);
 typedef ssize_t (*ngx_recv_chain_pt)(ngx_connection_t *c, ngx_chain_t *in);
 typedef ssize_t (*ngx_send_pt)(ngx_connection_t *c, u_char *buf, size_t size);
 typedef ngx_chain_t *(*ngx_send_chain_pt)(ngx_connection_t *c, ngx_chain_t *in,
-                                          off_t limit);
+    off_t limit);
 
 typedef struct {
     ngx_recv_pt        recv;
     ngx_recv_chain_pt  recv_chain;
+    ngx_recv_pt        udp_recv;
     ngx_send_pt        send;
     ngx_send_chain_pt  send_chain;
     ngx_uint_t         flags;
@@ -36,11 +37,14 @@ void ngx_os_status(ngx_log_t *log);
 
 ssize_t ngx_wsarecv(ngx_connection_t *c, u_char *buf, size_t size);
 ssize_t ngx_overlapped_wsarecv(ngx_connection_t *c, u_char *buf, size_t size);
+ssize_t ngx_udp_wsarecv(ngx_connection_t *c, u_char *buf, size_t size);
+ssize_t ngx_udp_overlapped_wsarecv(ngx_connection_t *c, u_char *buf,
+    size_t size);
 ssize_t ngx_wsarecv_chain(ngx_connection_t *c, ngx_chain_t *chain);
 ngx_chain_t *ngx_wsasend_chain(ngx_connection_t *c, ngx_chain_t *in,
-                               off_t limit);
+    off_t limit);
 ngx_chain_t *ngx_overlapped_wsasend_chain(ngx_connection_t *c, ngx_chain_t *in,
-                                          off_t limit);
+    off_t limit);
 
 
 extern ngx_os_io_t  ngx_os_io;

src/os/win32/ngx_udp_wsarecv.c

@@ -0,0 +1,146 @@
+
+/*
+ * Copyright (C) Igor Sysoev
+ */
+
+
+#include <ngx_config.h>
+#include <ngx_core.h>
+#include <ngx_event.h>
+
+
+ssize_t
+ngx_udp_wsarecv(ngx_connection_t *c, u_char *buf, size_t size)
+{
+    int           rc;
+    u_long        bytes, flags;
+    WSABUF        wsabuf[1];
+    ngx_err_t     err;
+    ngx_event_t  *rev;
+
+    wsabuf[0].buf = (char *) buf;
+    wsabuf[0].len = size;
+    flags = 0;
+    bytes = 0;
+
+    rc = WSARecv(c->fd, wsabuf, 1, &bytes, &flags, NULL, NULL);
+
+    ngx_log_debug4(NGX_LOG_DEBUG_EVENT, c->log, 0,
+                   "WSARecv: fd:%d rc:%d %ul of %z", c->fd, rc, bytes, size);
+
+    rev = c->read;
+
+    if (rc == -1) {
+        rev->ready = 0;
+        err = ngx_socket_errno;
+
+        if (err == WSAEWOULDBLOCK) {
+            ngx_log_debug0(NGX_LOG_DEBUG_EVENT, c->log, err,
+                           "WSARecv() not ready");
+            return NGX_AGAIN;
+        }
+
+        rev->error = 1;
+        ngx_connection_error(c, err, "WSARecv() failed");
+
+        return NGX_ERROR;
+    }
+
+    return bytes;
+}
+
+
+ssize_t
+ngx_udp_overlapped_wsarecv(ngx_connection_t *c, u_char *buf, size_t size)
+{
+    int               rc;
+    u_long            bytes, flags;
+    WSABUF            wsabuf[1];
+    ngx_err_t         err;
+    ngx_event_t      *rev;
+    LPWSAOVERLAPPED   ovlp;
+
+    rev = c->read;
+
+    if (!rev->ready) {
+        ngx_log_error(NGX_LOG_ALERT, c->log, 0, "second wsa post");
+        return NGX_AGAIN;
+    }
+
+    ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
+                   "rev->complete: %d", rev->complete);
+
+    if (rev->complete) {
+        rev->complete = 0;
+
+        if (ngx_event_flags & NGX_USE_IOCP_EVENT) {
+            if (rev->ovlp.error) {
+                ngx_connection_error(c, rev->ovlp.error, "WSARecv() failed");
+                return NGX_ERROR;
+            }
+
+            ngx_log_debug3(NGX_LOG_DEBUG_EVENT, c->log, 0,
+                           "WSARecv ovlp: fd:%d %ul of %z",
+                           c->fd, rev->available, size);
+
+            return rev->available;
+        }
+
+        if (WSAGetOverlappedResult(c->fd, (LPWSAOVERLAPPED) &rev->ovlp,
+                                   &bytes, 0, NULL) == 0) {
+            ngx_connection_error(c, ngx_socket_errno,
+                               "WSARecv() or WSAGetOverlappedResult() failed");
+            return NGX_ERROR;
+        }
+
+        ngx_log_debug3(NGX_LOG_DEBUG_EVENT, c->log, 0,
+                       "WSARecv: fd:%d %ul of %z", c->fd, bytes, size);
+
+        return bytes;
+    }
+
+    ovlp = (LPWSAOVERLAPPED) &rev->ovlp;
+    ngx_memzero(ovlp, sizeof(WSAOVERLAPPED));
+    wsabuf[0].buf = (char *) buf;
+    wsabuf[0].len = size;
+    flags = 0;
+    bytes = 0;
+
+    rc = WSARecv(c->fd, wsabuf, 1, &bytes, &flags, ovlp, NULL);
+
+    rev->complete = 0;
+
+    ngx_log_debug4(NGX_LOG_DEBUG_EVENT, c->log, 0,
+                   "WSARecv ovlp: fd:%d rc:%d %ul of %z",
+                   c->fd, rc, bytes, size);
+
+    if (rc == -1) {
+        err = ngx_socket_errno;
+        if (err == WSA_IO_PENDING) {
+            rev->active = 1;
+            ngx_log_debug0(NGX_LOG_DEBUG_EVENT, c->log, err,
+                           "WSARecv() posted");
+            return NGX_AGAIN;
+        }
+
+        rev->error = 1;
+        ngx_connection_error(c, err, "WSARecv() failed");
+        return NGX_ERROR;
+    }
+
+    if (ngx_event_flags & NGX_USE_IOCP_EVENT) {
+
+        /*
+         * if a socket was bound with I/O completion port
+         * then GetQueuedCompletionStatus() would anyway return its status
+         * despite that WSARecv() was already complete
+         */
+
+        rev->active = 1;
+        return NGX_AGAIN;
+    }
+
+    rev->active = 0;
+
+    return bytes;
+}

src/os/win32/ngx_win32_init.c

@@ -22,6 +22,7 @@ ngx_fd_t    ngx_stderr_fileno;
 ngx_os_io_t ngx_os_io = {
     ngx_wsarecv,
     ngx_wsarecv_chain,
+    ngx_udp_wsarecv,
     NULL,
     ngx_wsasend_chain,
     0