[PATCH 1 of 3] Stream: socket peek in preread phase

[Roman Arutyunyan]

# HG changeset patch
# User Roman Arutyunyan <arut at nginx.com>
# Date 1699456644 -14400
#      Wed Nov 08 19:17:24 2023 +0400
# Node ID 966331bb4936888ef2f034aa2700c130514d0b57
# Parent  7ec761f0365f418511e30b82e9adf80bc56681df
Stream: socket peek in preread phase.

Previously, preread buffer was always read out from socket, which made it
impossible to terminate SSL on the connection without introducing additional
SSL BIOs.  The following patches will rely on this.

Now, when possible, recv(MSG_PEEK) is used instead, which keeps data in socket.
It's called if SSL is not already terminated and if an egde-triggered event
method is used.  For epoll, EPOLLRDHUP support is also required.

diff --git a/src/stream/ngx_stream_core_module.c b/src/stream/ngx_stream_core_module.c
--- a/src/stream/ngx_stream_core_module.c
+++ b/src/stream/ngx_stream_core_module.c
@@ -10,6 +10,10 @@
 #include <ngx_stream.h>
 
 
+static ngx_int_t ngx_stream_preread_peek(ngx_stream_session_t *s,
+    ngx_stream_phase_handler_t *ph);
+static ngx_int_t ngx_stream_preread(ngx_stream_session_t *s,
+    ngx_stream_phase_handler_t *ph);
 static ngx_int_t ngx_stream_core_preconfiguration(ngx_conf_t *cf);
 static void *ngx_stream_core_create_main_conf(ngx_conf_t *cf);
 static char *ngx_stream_core_init_main_conf(ngx_conf_t *cf, void *conf);
@@ -203,8 +207,6 @@ ngx_int_t
 ngx_stream_core_preread_phase(ngx_stream_session_t *s,
     ngx_stream_phase_handler_t *ph)
 {
-    size_t                       size;
-    ssize_t                      n;
     ngx_int_t                    rc;
     ngx_connection_t            *c;
     ngx_stream_core_srv_conf_t  *cscf;
@@ -217,56 +219,40 @@ ngx_stream_core_preread_phase(ngx_stream
 
     if (c->read->timedout) {
         rc = NGX_STREAM_OK;
+        goto done;
+    }
 
-    } else if (c->read->timer_set) {
-        rc = NGX_AGAIN;
+    if (!c->read->timer_set) {
+        rc = ph->handler(s);
 
-    } else {
-        rc = ph->handler(s);
+        if (rc != NGX_AGAIN) {
+            goto done;
+        }
     }
 
-    while (rc == NGX_AGAIN) {
-
+    if (c->buffer == NULL) {
+        c->buffer = ngx_create_temp_buf(c->pool, cscf->preread_buffer_size);
         if (c->buffer == NULL) {
-            c->buffer = ngx_create_temp_buf(c->pool, cscf->preread_buffer_size);
-            if (c->buffer == NULL) {
-                rc = NGX_ERROR;
-                break;
-            }
+            rc = NGX_ERROR;
+            goto done;
         }
-
-        size = c->buffer->end - c->buffer->last;
-
-        if (size == 0) {
-            ngx_log_error(NGX_LOG_ERR, c->log, 0, "preread buffer full");
-            rc = NGX_STREAM_BAD_REQUEST;
-            break;
-        }
+    }
 
-        if (c->read->eof) {
-            rc = NGX_STREAM_OK;
-            break;
-        }
-
-        if (!c->read->ready) {
-            break;
-        }
-
-        n = c->recv(c, c->buffer->last, size);
+    if (c->ssl == NULL
+        && (ngx_event_flags & NGX_USE_CLEAR_EVENT)
+        && ((ngx_event_flags & NGX_USE_EPOLL_EVENT) == 0
+#if (NGX_HAVE_EPOLLRDHUP)
+            || ngx_use_epoll_rdhup
+#endif
+        ))
+    {
+        rc = ngx_stream_preread_peek(s, ph);
 
-        if (n == NGX_ERROR || n == 0) {
-            rc = NGX_STREAM_OK;
-            break;
-        }
+    } else {
+        rc = ngx_stream_preread(s, ph);
+    }
 
-        if (n == NGX_AGAIN) {
-            break;
-        }
-
-        c->buffer->last += n;
-
-        rc = ph->handler(s);
-    }
+done:
 
     if (rc == NGX_AGAIN) {
         if (ngx_handle_read_event(c->read, 0) != NGX_OK) {
@@ -311,6 +297,95 @@ ngx_stream_core_preread_phase(ngx_stream
 }
 
 
+static ngx_int_t
+ngx_stream_preread_peek(ngx_stream_session_t *s, ngx_stream_phase_handler_t *ph)
+{
+    ssize_t            n;
+    ngx_int_t          rc;
+    ngx_err_t          err;
+    ngx_connection_t  *c;
+
+    c = s->connection;
+
+    n = recv(c->fd, (char *) c->buffer->last,
+             c->buffer->end - c->buffer->last, MSG_PEEK);
+
+    err = ngx_socket_errno;
+
+    ngx_log_debug1(NGX_LOG_DEBUG_HTTP, c->log, 0,
+                   "stream recv(MSG_PEEK): %z", n);
+
+    if (n == -1) {
+        if (err == NGX_EAGAIN) {
+            return NGX_AGAIN;
+        }
+
+        ngx_connection_error(c, err, "recv() failed");
+        return NGX_STREAM_OK;
+    }
+
+    if (n == 0) {
+        return NGX_STREAM_OK;
+    }
+
+    c->buffer->last += n;
+
+    rc = ph->handler(s);
+
+    if (rc == NGX_AGAIN && c->buffer->last == c->buffer->end) {
+        ngx_log_error(NGX_LOG_ERR, c->log, 0, "preread buffer full");
+        return NGX_STREAM_BAD_REQUEST;
+    }
+
+    if (rc == NGX_AGAIN && c->read->pending_eof) {
+        return NGX_STREAM_OK;
+    }
+
+    c->buffer->last = c->buffer->pos;
+
+    return rc;
+}
+
+
+static ngx_int_t
+ngx_stream_preread(ngx_stream_session_t *s, ngx_stream_phase_handler_t *ph)
+{
+    ssize_t            n;
+    ngx_int_t          rc;
+    ngx_connection_t  *c;
+
+    c = s->connection;
+
+    while (c->read->ready) {
+
+        n = c->recv(c, c->buffer->last, c->buffer->end - c->buffer->last);
+
+        if (n == NGX_AGAIN) {
+            return NGX_AGAIN;
+        }
+
+        if (n == NGX_ERROR || n == 0) {
+            return NGX_STREAM_OK;
+        }
+
+        c->buffer->last += n;
+
+        rc = ph->handler(s);
+
+        if (rc != NGX_AGAIN) {
+            return rc;
+        }
+
+        if (c->buffer->last == c->buffer->end) {
+            ngx_log_error(NGX_LOG_ERR, c->log, 0, "preread buffer full");
+            return NGX_STREAM_BAD_REQUEST;
+        }
+    }
+
+    return NGX_AGAIN;
+}
+
+
 ngx_int_t
 ngx_stream_core_content_phase(ngx_stream_session_t *s,
     ngx_stream_phase_handler_t *ph)