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

[Sergey Kandaurov]

On Wed, Dec 13, 2023 at 06:06:59PM +0400, Roman Arutyunyan wrote:
> Hi,

> # HG changeset patch
> # User Roman Arutyunyan <arut at nginx.com>
> # Date 1702476295 -14400
> #      Wed Dec 13 18:04:55 2023 +0400
> # Node ID 844486cdd43a32d10b78493d7e7b80e9e2239d7e
> # Parent  6c8595b77e667bd58fd28186939ed820f2e55e0e
> 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,100 @@ 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_STREAM, c->log, 0, "stream recv(): %z", n);
> +
> +    if (n == -1) {
> +        if (err == NGX_EAGAIN) {
> +            c->read->ready = 0;
> +            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->pos;
> +        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;
> +    }
> +
> +    if (c->read->pending_eof) {
> +        return NGX_STREAM_OK;
> +    }
> +
> +    c->buffer->last = c->buffer->pos;
> +
> +    return NGX_AGAIN;
> +}
> +
> +
> +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)

Looks good.