Events: available bytes calculation via ioctl(FIONREAD).

This makes it possible to avoid looping for a long time while working
with a fast enough peer when data are added to the socket buffer faster
than we are able to read and process them (ticket #1431).  This is
basically what we already do on FreeBSD with kqueue, where information
about the number of bytes in the socket buffer is returned by
the kevent() call.

With other event methods rev->available is now set to -1 when the socket
is ready for reading.  Later in ngx_recv() and ngx_recv_chain(), if
full buffer is received, real number of bytes in the socket buffer is
retrieved using ioctl(FIONREAD).  Reading more than this number of bytes
ensures that even with edge-triggered event methods the event will be
triggered again, so it is safe to stop processing of the socket and
switch to other connections.

Using ioctl(FIONREAD) only after reading a full buffer is an optimization.
With this approach we only call ioctl(FIONREAD) when there are at least
two recv()/readv() calls.

diff --git a/auto/unix b/auto/unix
index 43d3b25a54563f8e9c905238f718b07400ffead6..ff9697a4eaa6203badb0ffb5464bdf4267d3ce3e 100644 (file)
--- a/auto/unix
+++ b/auto/unix
@@ -943,6 +943,18 @@ ngx_feature_test="int i = FIONBIO; printf(\"%d\", i)"
 . auto/feature
 
 
+ngx_feature="ioctl(FIONREAD)"
+ngx_feature_name="NGX_HAVE_FIONREAD"
+ngx_feature_run=no
+ngx_feature_incs="#include <sys/ioctl.h>
+                  #include <stdio.h>
+                  $NGX_INCLUDE_SYS_FILIO_H"
+ngx_feature_path=
+ngx_feature_libs=
+ngx_feature_test="int i = FIONREAD; printf(\"%d\", i)"
+. auto/feature
+
+
 ngx_feature="struct tm.tm_gmtoff"
 ngx_feature_name="NGX_HAVE_GMTOFF"
 ngx_feature_run=no

diff --git a/src/event/modules/ngx_devpoll_module.c b/src/event/modules/ngx_devpoll_module.c
index ee9f854b6f551f4acabe45c3ab1c5f9c58f49bb6..590eb28e14640a78c9a1670689aaab21cfd37954 100644 (file)
--- a/src/event/modules/ngx_devpoll_module.c
+++ b/src/event/modules/ngx_devpoll_module.c
@@ -495,6 +495,7 @@ ngx_devpoll_process_events(ngx_cycle_t *cycle, ngx_msec_t timer,
 
         if ((revents & POLLIN) && rev->active) {
             rev->ready = 1;
+            rev->available = -1;
 
             if (flags & NGX_POST_EVENTS) {
                 queue = rev->accept ? &ngx_posted_accept_events

diff --git a/src/event/modules/ngx_epoll_module.c b/src/event/modules/ngx_epoll_module.c
index 76aee08c390045055b1673c507eb8d7c3863640c..98e3ce7c826ffb23e29581109b61f45644ccdbe0 100644 (file)
--- a/src/event/modules/ngx_epoll_module.c
+++ b/src/event/modules/ngx_epoll_module.c
@@ -886,11 +886,10 @@ ngx_epoll_process_events(ngx_cycle_t *cycle, ngx_msec_t timer, ngx_uint_t flags)
             if (revents & EPOLLRDHUP) {
                 rev->pending_eof = 1;
             }
-
-            rev->available = 1;
 #endif
 
             rev->ready = 1;
+            rev->available = -1;
 
             if (flags & NGX_POST_EVENTS) {
                 queue = rev->accept ? &ngx_posted_accept_events

diff --git a/src/event/modules/ngx_eventport_module.c b/src/event/modules/ngx_eventport_module.c
index 11ad0934efe15721687e87d5481dfdfec7b0379f..f67c70457f493218117729c79e48a3c6ab603424 100644 (file)
--- a/src/event/modules/ngx_eventport_module.c
+++ b/src/event/modules/ngx_eventport_module.c
@@ -559,6 +559,7 @@ ngx_eventport_process_events(ngx_cycle_t *cycle, ngx_msec_t timer,
 
             if (revents & POLLIN) {
                 rev->ready = 1;
+                rev->available = -1;
 
                 if (flags & NGX_POST_EVENTS) {
                     queue = rev->accept ? &ngx_posted_accept_events

diff --git a/src/event/modules/ngx_poll_module.c b/src/event/modules/ngx_poll_module.c
index b46ab538abd6a39d695641535465a77df797b8b5..c16f0242e4a46f1e54da9b60e0168c95f40737c9 100644 (file)
--- a/src/event/modules/ngx_poll_module.c
+++ b/src/event/modules/ngx_poll_module.c
@@ -370,6 +370,7 @@ ngx_poll_process_events(ngx_cycle_t *cycle, ngx_msec_t timer, ngx_uint_t flags)
 
             ev = c->read;
             ev->ready = 1;
+            ev->available = -1;
 
             queue = ev->accept ? &ngx_posted_accept_events
                                : &ngx_posted_events;

diff --git a/src/event/modules/ngx_select_module.c b/src/event/modules/ngx_select_module.c
index 06446219f3aaeeebacc010fa91a04a7e443e74c2..b9fceb3b648be2f45a3ad5f293d7c2c78bfa23c4 100644 (file)
--- a/src/event/modules/ngx_select_module.c
+++ b/src/event/modules/ngx_select_module.c
@@ -330,6 +330,7 @@ ngx_select_process_events(ngx_cycle_t *cycle, ngx_msec_t timer,
 
         if (found) {
             ev->ready = 1;
+            ev->available = -1;
 
             queue = ev->accept ? &ngx_posted_accept_events
                                : &ngx_posted_events;

diff --git a/src/event/modules/ngx_win32_poll_module.c b/src/event/modules/ngx_win32_poll_module.c
index 9fe867f7904abb9584750b05638ffaf4c33f2a46..2fbc1b376aef3040ec43ea16e5b9bc02dffe0cd4 100644 (file)
--- a/src/event/modules/ngx_win32_poll_module.c
+++ b/src/event/modules/ngx_win32_poll_module.c
@@ -380,6 +380,7 @@ ngx_poll_process_events(ngx_cycle_t *cycle, ngx_msec_t timer, ngx_uint_t flags)
 
             ev = c->read;
             ev->ready = 1;
+            ev->available = -1;
 
             queue = ev->accept ? &ngx_posted_accept_events
                                : &ngx_posted_events;

diff --git a/src/event/modules/ngx_win32_select_module.c b/src/event/modules/ngx_win32_select_module.c
index 8093a600b0e4a0d053bb6dc404582203b8bd0cc9..962514a3668fb4e6aa9be9a454c4a6b4c43c8ce8 100644 (file)
--- a/src/event/modules/ngx_win32_select_module.c
+++ b/src/event/modules/ngx_win32_select_module.c
@@ -330,6 +330,7 @@ ngx_select_process_events(ngx_cycle_t *cycle, ngx_msec_t timer,
 
         if (found) {
             ev->ready = 1;
+            ev->available = -1;
 
             queue = ev->accept ? &ngx_posted_accept_events
                                : &ngx_posted_events;

diff --git a/src/event/ngx_event.h b/src/event/ngx_event.h
index bb77c4ae664d8357c3bb1bb39a4385eef13503d9..97f9673c90dac71c00ab75a08108438b3cd4febd 100644 (file)
--- a/src/event/ngx_event.h
+++ b/src/event/ngx_event.h
@@ -91,21 +91,14 @@ struct ngx_event_s {
      *   write:      available space in buffer when event is ready
      *               or lowat when event is set with NGX_LOWAT_EVENT flag
      *
-     * epoll with EPOLLRDHUP:
-     *   accept:     1 if accept many, 0 otherwise
-     *   read:       1 if there can be data to read, 0 otherwise
-     *
      * iocp: TODO
      *
      * otherwise:
      *   accept:     1 if accept many, 0 otherwise
+     *   read:       bytes to read when event is ready, -1 if not known
      */
 
-#if (NGX_HAVE_KQUEUE) || (NGX_HAVE_IOCP)
     int              available;
-#else
-    unsigned         available:1;
-#endif
 
     ngx_event_handler_pt  handler;
 

diff --git a/src/os/unix/ngx_readv_chain.c b/src/os/unix/ngx_readv_chain.c
index 454cfdcad44e2c4569c2c639ac711111d73d4700..a3577ce19212dc071d5abdf4f4977640bf77af2f 100644 (file)
--- a/src/os/unix/ngx_readv_chain.c
+++ b/src/os/unix/ngx_readv_chain.c
@@ -60,7 +60,7 @@ ngx_readv_chain(ngx_connection_t *c, ngx_chain_t *chain, off_t limit)
                        "readv: eof:%d, avail:%d",
                        rev->pending_eof, rev->available);
 
-        if (!rev->available && !rev->pending_eof) {
+        if (rev->available == 0 && !rev->pending_eof) {
             return NGX_AGAIN;
         }
     }
@@ -165,6 +165,40 @@ ngx_readv_chain(ngx_connection_t *c, ngx_chain_t *chain, off_t limit)
 
 #endif
 
+#if (NGX_HAVE_FIONREAD)
+
+            if (rev->available >= 0) {
+                rev->available -= n;
+
+                /*
+                 * negative rev->available means some additional bytes
+                 * were received between kernel notification and readv(),
+                 * and therefore ev->ready can be safely reset even for
+                 * edge-triggered event methods
+                 */
+
+                if (rev->available < 0) {
+                    rev->available = 0;
+                    rev->ready = 0;
+                }
+
+                ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
+                               "readv: avail:%d", rev->available);
+
+            } else if (n == size) {
+
+                if (ngx_socket_nread(c->fd, &rev->available) == -1) {
+                    n = ngx_connection_error(c, ngx_socket_errno,
+                                             ngx_socket_nread_n " failed");
+                    break;
+                }
+
+                ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
+                               "readv: avail:%d", rev->available);
+            }
+
+#endif
+
 #if (NGX_HAVE_EPOLLRDHUP)
 
             if ((ngx_event_flags & NGX_USE_EPOLL_EVENT)

diff --git a/src/os/unix/ngx_recv.c b/src/os/unix/ngx_recv.c
index c85fd453c91bfffd842f5b73863d453865ebab9c..ddfae4dbee66069043bb7149d6686889a3b2ef1e 100644 (file)
--- a/src/os/unix/ngx_recv.c
+++ b/src/os/unix/ngx_recv.c
@@ -57,7 +57,7 @@ ngx_unix_recv(ngx_connection_t *c, u_char *buf, size_t size)
                        "recv: eof:%d, avail:%d",
                        rev->pending_eof, rev->available);
 
-        if (!rev->available && !rev->pending_eof) {
+        if (rev->available == 0 && !rev->pending_eof) {
             rev->ready = 0;
             return NGX_AGAIN;
         }
@@ -116,6 +116,40 @@ ngx_unix_recv(ngx_connection_t *c, u_char *buf, size_t size)
 
 #endif
 
+#if (NGX_HAVE_FIONREAD)
+
+            if (rev->available >= 0) {
+                rev->available -= n;
+
+                /*
+                 * negative rev->available means some additional bytes
+                 * were received between kernel notification and recv(),
+                 * and therefore ev->ready can be safely reset even for
+                 * edge-triggered event methods
+                 */
+
+                if (rev->available < 0) {
+                    rev->available = 0;
+                    rev->ready = 0;
+                }
+
+                ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
+                               "recv: avail:%d", rev->available);
+
+            } else if ((size_t) n == size) {
+
+                if (ngx_socket_nread(c->fd, &rev->available) == -1) {
+                    n = ngx_connection_error(c, ngx_socket_errno,
+                                             ngx_socket_nread_n " failed");
+                    break;
+                }
+
+                ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
+                               "recv: avail:%d", rev->available);
+            }
+
+#endif
+
 #if (NGX_HAVE_EPOLLRDHUP)
 
             if ((ngx_event_flags & NGX_USE_EPOLL_EVENT)

diff --git a/src/os/unix/ngx_socket.h b/src/os/unix/ngx_socket.h
index fcc51533568e1abb744a5502c4b738c15f41ac60..ec66a6f830f8a70c88046ed37ee82c59137e3154 100644 (file)
--- a/src/os/unix/ngx_socket.h
+++ b/src/os/unix/ngx_socket.h
@@ -38,6 +38,13 @@ int ngx_blocking(ngx_socket_t s);
 
 #endif
 
+#if (NGX_HAVE_FIONREAD)
+
+#define ngx_socket_nread(s, n)  ioctl(s, FIONREAD, n)
+#define ngx_socket_nread_n      "ioctl(FIONREAD)"
+
+#endif
+
 int ngx_tcp_nopush(ngx_socket_t s);
 int ngx_tcp_push(ngx_socket_t s);
 

diff --git a/src/os/win32/ngx_socket.c b/src/os/win32/ngx_socket.c
index 05a39f4cc7be2ae0a30ef3b620dbd7722f33308c..b1b4afb8a0eeae43c846a58db61994d5de011dda 100644 (file)
--- a/src/os/win32/ngx_socket.c
+++ b/src/os/win32/ngx_socket.c
@@ -27,6 +27,21 @@ ngx_blocking(ngx_socket_t s)
 }
 
 
+int
+ngx_socket_nread(ngx_socket_t s, int *n)
+{
+    unsigned long  nread;
+
+    if (ioctlsocket(s, FIONREAD, &nread) == -1) {
+        return -1;
+    }
+
+    *n = nread;
+
+    return 0;
+}
+
+
 int
 ngx_tcp_push(ngx_socket_t s)
 {

diff --git a/src/os/win32/ngx_socket.h b/src/os/win32/ngx_socket.h
index f8a453d562bfb4615a36b13e23d1b8a63b2338f1..ab56bc8b34af55552d69ce82f2943149f941f81b 100644 (file)
--- a/src/os/win32/ngx_socket.h
+++ b/src/os/win32/ngx_socket.h
@@ -31,6 +31,9 @@ int ngx_blocking(ngx_socket_t s);
 #define ngx_nonblocking_n   "ioctlsocket(FIONBIO)"
 #define ngx_blocking_n      "ioctlsocket(!FIONBIO)"
 
+int ngx_socket_nread(ngx_socket_t s, int *n);
+#define ngx_socket_nread_n  "ioctlsocket(FIONREAD)"
+
 #define ngx_shutdown_socket    shutdown
 #define ngx_shutdown_socket_n  "shutdown()"
 

diff --git a/src/os/win32/ngx_win32_config.h b/src/os/win32/ngx_win32_config.h
index 4824d05a8759a3ca61391205fc27022349eb9cf4..96156870d0a6f533e74b2b921dd671d626ad5ba7 100644 (file)
--- a/src/os/win32/ngx_win32_config.h
+++ b/src/os/win32/ngx_win32_config.h
@@ -273,6 +273,10 @@ typedef int                 sig_atomic_t;
 #define NGX_HAVE_SO_SNDLOWAT         0
 #endif
 
+#ifndef NGX_HAVE_FIONREAD
+#define NGX_HAVE_FIONREAD            1
+#endif
+
 #define NGX_HAVE_GETADDRINFO         1
 
 #define ngx_random               rand

diff --git a/src/os/win32/ngx_wsarecv.c b/src/os/win32/ngx_wsarecv.c
index 1925f0b17ed178ff60b42fd7ec8f647615d7a93b..ac883107b14991d1f5ca08231730770419ab1d35 100644 (file)
--- a/src/os/win32/ngx_wsarecv.c
+++ b/src/os/win32/ngx_wsarecv.c
@@ -51,6 +51,45 @@ ngx_wsarecv(ngx_connection_t *c, u_char *buf, size_t size)
         return n;
     }
 
+#if (NGX_HAVE_FIONREAD)
+
+    if (rev->available >= 0 && bytes > 0) {
+        rev->available -= bytes;
+
+        /*
+         * negative rev->available means some additional bytes
+         * were received between kernel notification and WSARecv(),
+         * and therefore ev->ready can be safely reset even for
+         * edge-triggered event methods
+         */
+
+        if (rev->available < 0) {
+            rev->available = 0;
+            rev->ready = 0;
+        }
+
+        ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
+                       "WSARecv: avail:%d", rev->available);
+
+    } else if (bytes == size) {
+
+        if (ngx_socket_nread(c->fd, &rev->available) == -1) {
+            n = ngx_connection_error(c, ngx_socket_errno,
+                                     ngx_socket_nread_n " failed");
+
+            if (n == NGX_ERROR) {
+                rev->error = 1;
+            }
+
+            return n;
+        }
+
+        ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
+                       "WSARecv: avail:%d", rev->available);
+    }
+
+#endif
+
     if (bytes < size) {
         rev->ready = 0;
     }

diff --git a/src/os/win32/ngx_wsarecv_chain.c b/src/os/win32/ngx_wsarecv_chain.c
index 2598e091c37ec786a0e5fbe1ce291bdcbeaf1951..87f02391152a2280e1913e4462fc3aa4b70064a0 100644 (file)
--- a/src/os/win32/ngx_wsarecv_chain.c
+++ b/src/os/win32/ngx_wsarecv_chain.c
@@ -94,6 +94,41 @@ ngx_wsarecv_chain(ngx_connection_t *c, ngx_chain_t *chain, off_t limit)
         return NGX_ERROR;
     }
 
+#if (NGX_HAVE_FIONREAD)
+
+    if (rev->available >= 0 && bytes > 0) {
+        rev->available -= bytes;
+
+        /*
+         * negative rev->available means some additional bytes
+         * were received between kernel notification and WSARecv(),
+         * and therefore ev->ready can be safely reset even for
+         * edge-triggered event methods
+         */
+
+        if (rev->available < 0) {
+            rev->available = 0;
+            rev->ready = 0;
+        }
+
+        ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
+                       "WSARecv: avail:%d", rev->available);
+
+    } else if (bytes == size) {
+
+        if (ngx_socket_nread(c->fd, &rev->available) == -1) {
+            rev->error = 1;
+            ngx_connection_error(c, ngx_socket_errno,
+                                 ngx_socket_nread_n " failed");
+            return NGX_ERROR;
+        }
+
+        ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
+                       "WSARecv: avail:%d", rev->available);
+    }
+
+#endif
+
     if (bytes < size) {
         rev->ready = 0;
     }