--- a/src/event/ngx_event_pipe.c
+++ b/src/event/ngx_event_pipe.c
@@ -19,6 +19,8 @@ static int ngx_event_pipe_drain_chains(n
 
 int ngx_event_pipe(ngx_event_pipe_t *p, int do_write)
 {
+    ngx_event_t  *rev, *wev;
+
     for ( ;; ) {
         if (do_write) {
             if (ngx_event_pipe_write_to_downstream(p) == NGX_ABORT) {
@@ -40,15 +42,26 @@ int ngx_event_pipe(ngx_event_pipe_t *p, 
         do_write = 1;
     }
 
-    if (ngx_handle_read_event(p->upstream->read) == NGX_ERROR) {
+    rev = p->upstream->read;
+
+    if (ngx_handle_read_event(rev, (rev->eof || rev->error)) == NGX_ERROR) {
         return NGX_ABORT;
     }
 
-    if (ngx_handle_write_event(p->downstream->write,
-                                           /* TODO: lowat */ 0) == NGX_ERROR) {
+    if (rev->active) {
+        ngx_add_timer(rev, p->read_timeout);
+    }
+
+    wev = p->downstream->write;
+
+    if (ngx_handle_write_event(wev, p->send_lowat) == NGX_ERROR) {
         return NGX_ABORT;
     }
 
+    if (wev->active) {
+        ngx_add_timer(wev, p->send_timeout);
+    }
+
     return NGX_OK;
 }
 
@@ -112,7 +125,6 @@ int ngx_event_pipe_read_upstream(ngx_eve
 
                             ngx_log_error(NGX_LOG_ERR, p->log,
                                           p->upstream->read->kq_errno,
-                                          /* TODO: ngx_readv_chain_n */
                                           "readv() failed");
                         }
 
@@ -213,8 +225,6 @@ int ngx_event_pipe_read_upstream(ngx_eve
                 break;
             }
 
-            ngx_log_debug(p->log, "HUNK_FREE: %d" _ chain->hunk->num);
-
             n = ngx_recv_chain(p->upstream, chain);
 
             ngx_log_debug(p->log, "recv_chain: %d" _ n);
@@ -343,8 +353,6 @@ int ngx_event_pipe_write_to_downstream(n
                 ngx_event_pipe_free_shadow_raw_hunk(&p->free_raw_hunks,
                                                     cl->hunk);
 
-ngx_log_debug(p->log, "HUNK OUT: %d %x" _ cl->hunk->num _ cl->hunk->type);
-
             } else if (!p->cachable && p->in) {
                 cl = p->in;
 
@@ -356,8 +364,6 @@ ngx_log_debug(p->log, "HUNK OUT: %d %x" 
 
                 p->in = p->in->next;
 
-ngx_log_debug(p->log, "HUNK IN: %d" _ cl->hunk->num);
-
             } else {
                 break;
             }