nginx-0.0.1-2003-11-18-11:04:34 import

src/event/ngx_event_pipe.c

@@ -54,6 +54,8 @@ int ngx_event_pipe(ngx_event_pipe_t *p, int do_write)
         }
     }
 
+ngx_log_debug(p->log, "DOWN: %d" _ p->downstream->fd);
+
     if (p->downstream->fd != -1) {
         wev = p->downstream->write;
 
@@ -673,12 +675,15 @@ static int ngx_event_pipe_drain_chains(ngx_event_pipe_t *p)
     for ( ;; ) {
         if (p->busy) {
             cl = p->busy;
+            p->busy = NULL;
 
         } else if (p->out) {
             cl = p->out;
+            p->out = NULL;
 
         } else if (p->in) {
             cl = p->in;
+            p->in = NULL;
 
         } else {
             return NGX_OK;

src/http/modules/proxy/ngx_http_proxy_upstream.c

@@ -19,6 +19,7 @@ static void ngx_http_proxy_process_upstream_status_line(ngx_event_t *rev);
 static void ngx_http_proxy_process_upstream_headers(ngx_event_t *rev);
 static ssize_t ngx_http_proxy_read_upstream_header(ngx_http_proxy_ctx_t *);
 static void ngx_http_proxy_send_response(ngx_http_proxy_ctx_t *p);
+static void ngx_http_proxy_check_broken_connection(ngx_event_t *wev);
 static void ngx_http_proxy_process_body(ngx_event_t *ev);
 static void ngx_http_proxy_next_upstream(ngx_http_proxy_ctx_t *p, int ft_type);
 
@@ -224,6 +225,23 @@ ngx_log_debug(r->connection->log, "timer_set: %d" _
         ngx_del_timer(r->connection->read);
     }
 
+    if ((ngx_event_flags & (NGX_USE_CLEAR_EVENT|NGX_HAVE_KQUEUE_EVENT))
+        && !r->connection->write->active)
+    {
+        /* kqueue allows to detect when client closes prematurely connection */
+
+        r->connection->write->event_handler =
+                                        ngx_http_proxy_check_broken_connection;
+
+        if (ngx_add_event(r->connection->write, NGX_WRITE_EVENT,
+                                                NGX_CLEAR_EVENT) == NGX_ERROR)
+        {
+            ngx_http_finalize_request(r, NGX_HTTP_INTERNAL_SERVER_ERROR);
+            return;
+        }
+    }
+
+
     if (!(cl = ngx_http_proxy_create_request(p))) {
         ngx_http_finalize_request(r, NGX_HTTP_INTERNAL_SERVER_ERROR);
         return;
@@ -424,7 +442,7 @@ static void ngx_http_proxy_connect(ngx_http_proxy_ctx_t *p)
 
     /* rc == NGX_OK */
 
-#if 1 /* test only */
+#if 1 /* test only, see below about "post aio operation" */
 
     if (c->read->ready) {
         /* post aio operation */
@@ -548,6 +566,11 @@ static void ngx_http_proxy_send_request_handler(ngx_event_t *wev)
         return;
     }
 
+    if (p->request->connection->write->eof) {
+        ngx_http_proxy_close_connection(p);
+        ngx_http_close_connection(p->request->connection);
+    }
+
     ngx_http_proxy_send_request(p);
 }
 
@@ -939,12 +962,16 @@ static void ngx_http_proxy_send_response(ngx_http_proxy_ctx_t *p)
         return;
     }
 
+    ep->cachable = p->cachable;
+
     ep->temp_file->file.fd = NGX_INVALID_FILE;
     ep->temp_file->file.log = r->connection->log;
     ep->temp_file->path = p->lcf->temp_path;
     ep->temp_file->pool = r->pool;
-    ep->temp_file->warn = "an upstream response is buffered "
+    if (!p->cachable) {
-                         "to a temporary file";
+        ep->temp_file->warn = "an upstream response is buffered "
+                              "to a temporary file";
+    }
 
     ep->max_temp_file_size = p->lcf->max_temp_file_size;
     ep->temp_file_write_size = p->lcf->temp_file_write_size;
@@ -983,8 +1010,6 @@ static void ngx_http_proxy_send_response(ngx_http_proxy_ctx_t *p)
      */
     p->header_in->last = p->header_in->pos;
 
-    ep->cachable = p->cachable;
-
     if (p->lcf->cyclic_temp_file) {
 
         /*
@@ -1017,6 +1042,45 @@ static void ngx_http_proxy_send_response(ngx_http_proxy_ctx_t *p)
 }
 
 
+static void ngx_http_proxy_check_broken_connection(ngx_event_t *wev)
+{
+    ngx_connection_t      *c;
+    ngx_http_request_t    *r;
+    ngx_http_proxy_ctx_t  *p;
+
+    ngx_log_debug(wev->log, "http proxy check client");
+
+    c = wev->data;
+    r = c->data;
+    p = ngx_http_get_module_ctx(r, ngx_http_proxy_module);
+
+#if (HAVE_KQUEUE)
+    if (wev->kq_eof) {
+        wev->eof = 1;
+
+        if (wev->kq_errno) {
+            wev->error = 1;
+        }
+
+        if (!p->cachable && p->upstream->peer.connection) {
+            ngx_log_error(NGX_LOG_INFO, wev->log, wev->kq_errno,
+                          "client closed prematurely connection, "
+                          "so upstream connection is closed too");
+            ngx_http_proxy_close_connection(p);
+
+        } else {
+            ngx_log_error(NGX_LOG_INFO, wev->log, wev->kq_errno,
+                          "client closed prematurely connection");
+        }
+ 
+        if (p->upstream->peer.connection == NULL) {
+            ngx_http_close_connection(c);
+        }
+    }
+#endif
+}
+
+
 static void ngx_http_proxy_process_body(ngx_event_t *ev)
 {
     ngx_connection_t      *c;
@@ -1097,11 +1161,9 @@ static void ngx_http_proxy_process_body(ngx_event_t *ev)
         }
  
         if (p->upstream->peer.connection == NULL) {
-            ngx_http_close_connection(c);
+            ngx_http_close_connection(r->connection);
         }
     }
-
-    return;
 }
 
 

src/os/unix/ngx_readv_chain.c

@@ -4,6 +4,10 @@
 #include <ngx_event.h>
 
 
+static int ngx_readv_error(ngx_event_t *rev, ngx_err_t err);
+
+#if (HAVE_KQUEUE)
+
 ssize_t ngx_readv_chain(ngx_connection_t *c, ngx_chain_t *chain)
 {
     char          *prev;
@@ -11,6 +15,32 @@ ssize_t ngx_readv_chain(ngx_connection_t *c, ngx_chain_t *chain)
     struct iovec  *iov;
     ngx_err_t      err;
     ngx_array_t    io;
+    ngx_event_t   *rev;
+
+    rev = c->read; 
+
+    if (ngx_event_flags & NGX_HAVE_KQUEUE_EVENT) {
+        ngx_log_debug(c->log, "recv: eof:%d, avail:%d, err:%d" _
+                      rev->kq_eof _ rev->available _ rev->kq_errno);
+
+        if (rev->available == 0) {
+            if (rev->kq_eof) {
+                rev->ready = 0;
+                rev->eof = 1;
+
+                if (rev->kq_errno) {
+                    rev->error = 1;
+                    ngx_set_socket_errno(rev->kq_errno);
+                    return ngx_readv_error(rev, rev->kq_errno);
+                }
+
+                return 0;
+
+            } else {
+                return NGX_AGAIN;
+            }
+        }
+    }
 
     prev = NULL;
     iov = NULL;
@@ -37,27 +67,143 @@ ssize_t ngx_readv_chain(ngx_connection_t *c, ngx_chain_t *chain)
 
 ngx_log_debug(c->log, "recv: %d:%d" _ io.nelts _ iov->iov_len);
 
-    n = readv(c->fd, (struct iovec *) io.elts, io.nelts);
+    rev = c->read;
 
-    if (n == 0) {
+    do {
-        c->read->eof = 1;
+        n = readv(c->fd, (struct iovec *) io.elts, io.nelts);
 
-    } else if (n == -1) {
+        if (n >= 0) {
-        c->read->ready = 0;
+            if (ngx_event_flags & NGX_HAVE_KQUEUE_EVENT) {
+                rev->available -= n;
 
-        err = ngx_errno;
+                /*
-        if (err == NGX_EAGAIN) {
+                 * rev->available can be negative here because some additional
-            ngx_log_error(NGX_LOG_INFO, c->log, err, "readv() returned EAGAIN");
+                 * bytes can be received between kevent() and recv()
-            return NGX_AGAIN;
+                 */
+
+                if (rev->available <= 0) {
+                    if (!rev->kq_eof) {
+                        rev->ready = 0;
+                    }
+
+                    if (rev->available < 0) {
+                        rev->available = 0;
+                    }
+                }
+
+                return n;
+            }
+
+            if (n < size) {
+                rev->ready = 0;
+            }
+
+            if (n == 0) {
+                rev->eof = 1;
+            }
+
+            return n;
         }
 
-        c->read->error = 1;
+        n = ngx_readv_error(rev, ngx_socket_errno);
-        ngx_log_error(NGX_LOG_ERR, c->log, err, "readv() failed");
-        return NGX_ERROR;
 
-    } else if (n < size) {
+    } while (n == NGX_EINTR);
-        c->read->ready = 0;
+
+    /* NGX_ERROR || NGX_AGAIN */
+
+    rev->ready = 0;
+
+    if (n == NGX_ERROR){
+        c->read->error = 1;
     }
 
     return n;
 }
+
+#else /* ! NAVE_KQUEUE */
+
+ssize_t ngx_readv_chain(ngx_connection_t *c, ngx_chain_t *chain)
+{
+    char          *prev;
+    ssize_t        n, size;
+    struct iovec  *iov;
+    ngx_err_t      err;
+    ngx_array_t    io;
+    ngx_event_t   *rev;
+
+    prev = NULL;
+    iov = NULL;
+    size = 0;
+
+    ngx_init_array(io, c->pool, 10, sizeof(struct iovec), NGX_ERROR);
+
+    /* coalesce the neighbouring hunks */
+
+    while (chain) {
+        if (prev == chain->hunk->last) {
+            iov->iov_len += chain->hunk->end - chain->hunk->last;
+
+        } else {
+            ngx_test_null(iov, ngx_push_array(&io), NGX_ERROR);
+            iov->iov_base = chain->hunk->last;
+            iov->iov_len = chain->hunk->end - chain->hunk->last;
+        }
+
+        size += chain->hunk->end - chain->hunk->last;
+        prev = chain->hunk->end;
+        chain = chain->next;
+    }
+
+ngx_log_debug(c->log, "recv: %d:%d" _ io.nelts _ iov->iov_len);
+
+    rev = c->read;
+
+    do {
+        n = readv(c->fd, (struct iovec *) io.elts, io.nelts);
+
+        if (n >= 0) {
+            if (n < size) {
+                rev->ready = 0;
+            }
+
+            if (n == 0) {
+                rev->eof = 1;
+            }
+
+            return n;
+        }
+
+        n = ngx_readv_error(rev, ngx_socket_errno);
+
+    } while (n == NGX_EINTR);
+
+    /* NGX_ERROR || NGX_AGAIN */
+
+    rev->ready = 0;
+
+    if (n == NGX_ERROR){
+        c->read->error = 1;
+    }
+
+    return n;
+}
+
+#endif /* NAVE_KQUEUE */
+
+
+static int ngx_readv_error(ngx_event_t *rev, ngx_err_t err)
+{
+    if (err == NGX_EAGAIN) {
+        ngx_log_error(NGX_LOG_INFO, rev->log, err, "readv() returned EAGAIN");
+        return NGX_AGAIN;
+    }
+
+    if (err == NGX_EINTR) {
+        ngx_log_error(NGX_LOG_INFO, rev->log, err, "readv() returned EINTR");
+        return NGX_EINTR;
+    }
+
+    ngx_log_error(NGX_LOG_ERR, rev->log, err, "readv() failed");
+
+    return NGX_ERROR;
+}

src/os/unix/ngx_recv.c

@@ -40,8 +40,6 @@ ssize_t ngx_unix_recv(ngx_connection_t *c, char *buf, size_t size)
     }
 
     do {
-        rev->ready = 1;
-
         n = recv(c->fd, buf, size, 0);
 
         ngx_log_debug(c->log, "recv: %d:%d" _ n _ size);
@@ -79,11 +77,14 @@ ssize_t ngx_unix_recv(ngx_connection_t *c, char *buf, size_t size)
             return n;
         }
 
-        rev->ready = 0;
         n = ngx_unix_recv_error(rev, ngx_socket_errno);
 
     } while (n == NGX_EINTR);
 
+    /* NGX_ERROR || NGX_AGAIN */
+
+    rev->ready = 0;
+
     if (n == NGX_ERROR){
         rev->error = 1;
     }
@@ -101,8 +102,6 @@ ssize_t ngx_unix_recv(ngx_connection_t *c, char *buf, size_t size)
     rev = c->read;
 
     do {
-        rev->ready = 1;
-
         n = recv(c->fd, buf, size, 0);
 
         ngx_log_debug(c->log, "recv: %d:%d" _ n _ size);
@@ -119,11 +118,14 @@ ssize_t ngx_unix_recv(ngx_connection_t *c, char *buf, size_t size)
             return n;
         }
 
-        rev->ready = 0;
         n = ngx_unix_recv_error(rev, ngx_socket_errno);
 
     } while (n == NGX_EINTR);
 
+    /* NGX_ERROR || NGX_AGAIN */
+
+    rev->ready = 0;
+
     if (n == NGX_ERROR){
         rev->error = 1;
     }