From e67d46189c911ca7dbeef8c7b7c99c4751683948 Mon Sep 17 00:00:00 2001 From: Igor Sysoev Date: Mon, 3 Dec 2007 16:46:46 +0000 Subject: [PATCH] ngx_udp_recv() --- auto/sources | 2 + src/core/ngx_resolver.c | 4 +- src/event/modules/ngx_aio_module.c | 1 + src/event/modules/ngx_iocp_module.c | 1 + src/event/ngx_event.h | 1 + src/os/unix/ngx_freebsd_init.c | 1 + src/os/unix/ngx_linux_init.c | 1 + src/os/unix/ngx_os.h | 2 + src/os/unix/ngx_posix_init.c | 1 + src/os/unix/ngx_solaris_init.c | 1 + src/os/unix/ngx_udp_recv.c | 114 ++++++++++++++++++++++ src/os/win32/ngx_os.h | 10 +- src/os/win32/ngx_udp_wsarecv.c | 146 ++++++++++++++++++++++++++++ src/os/win32/ngx_win32_init.c | 1 + 14 files changed, 281 insertions(+), 5 deletions(-) create mode 100644 src/os/unix/ngx_udp_recv.c create mode 100644 src/os/win32/ngx_udp_wsarecv.c diff --git a/auto/sources b/auto/sources index 074f96478..788d394a2 100644 --- a/auto/sources +++ b/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 \ diff --git a/src/core/ngx_resolver.c b/src/core/ngx_resolver.c index 6963acd02..ab4dcdabc 100644 --- a/src/core/ngx_resolver.c +++ b/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; } diff --git a/src/event/modules/ngx_aio_module.c b/src/event/modules/ngx_aio_module.c index 4a78c3991..985b065c6 100644 --- a/src/event/modules/ngx_aio_module.c +++ b/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 diff --git a/src/event/modules/ngx_iocp_module.c b/src/event/modules/ngx_iocp_module.c index 576ce1aae..837517151 100644 --- a/src/event/modules/ngx_iocp_module.c +++ b/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 diff --git a/src/event/ngx_event.h b/src/event/ngx_event.h index ed6ba29c1..a96b8bab9 100644 --- a/src/event/ngx_event.h +++ b/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 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 +#include +#include + + +#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 +#include +#include + + +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