ngx_udp_recv()

9 files changed, 274 insertions, 3 deletions

diff --git a/src/os/unix/ngx_freebsd_init.c b/src/os/unix/ngx_freebsd_init.c
index cb4482947..26499bd32 100644
--- a/src/os/unix/ngx_freebsd_init.c
+++ b/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,
diff --git a/src/os/unix/ngx_linux_init.c b/src/os/unix/ngx_linux_init.c
index f372e82a3..cb83cd3e8 100644
--- a/src/os/unix/ngx_linux_init.c
+++ b/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,
diff --git a/src/os/unix/ngx_os.h b/src/os/unix/ngx_os.h
index 015d64012..54b1370bd 100644
--- a/src/os/unix/ngx_os.h
+++ b/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);
diff --git a/src/os/unix/ngx_posix_init.c b/src/os/unix/ngx_posix_init.c
index 39e1b584a..78c5d4be9 100644
--- a/src/os/unix/ngx_posix_init.c
+++ b/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
diff --git a/src/os/unix/ngx_solaris_init.c b/src/os/unix/ngx_solaris_init.c
index a25505974..57a859e68 100644
--- a/src/os/unix/ngx_solaris_init.c
+++ b/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,
diff --git a/src/os/unix/ngx_udp_recv.c b/src/os/unix/ngx_udp_recv.c
new file mode 100644
index 000000000..25f0738ac
--- /dev/null
+++ b/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 */
diff --git a/src/os/win32/ngx_os.h b/src/os/win32/ngx_os.h
index 611f7ef5c..c9a911a75 100644
--- a/src/os/win32/ngx_os.h
+++ b/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;
diff --git a/src/os/win32/ngx_udp_wsarecv.c b/src/os/win32/ngx_udp_wsarecv.c
new file mode 100644
index 000000000..017fc6071
--- /dev/null
+++ b/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;
+}
diff --git a/src/os/win32/ngx_win32_init.c b/src/os/win32/ngx_win32_init.c
index 4437d338d..5ceb7a84b 100644
--- a/src/os/win32/ngx_win32_init.c
+++ b/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