[pve-devel] [RFC PATCH] fix #3106: correctly queue incoming connections

[Dominik Csapak]

looks better than mine :)

comment inline

On 11/3/20 1:26 PM, Dietmar Maurer wrote:
> ---
> 
>   based, on Domink's patch, but with the following changes:
> 
>   - factor out code into separate function accept_connections()
>   - no select with shutdown future (no needed)
>   - remove sender2.send_timeout() - not sure why this was there?
>   - restict number of spawned tasks
> 
>   Seems to work, but I get many handshake errors when connetion
>   with the GUI:
> 
>   > https handshake failed - the handshake failed: unexpected EOF
> 
>   This is because of pve status ping (Thomas will fix that in pve)
> 
>   But I am not sure why I get the following?
>   
>   > https handshakeX failed - the handshake failed: error:14094416:SSL routines:ssl3_read_bytes:sslv3 alert certificate unknown:../ssl/record/rec_layer_s3.c:1544:SSL alert number 46

i got that too, my guess was that the browser tries with client 
certificates?

> 
> 
>   src/bin/proxmox-backup-proxy.rs | 81 ++++++++++++++++++++++++++-------
>   1 file changed, 64 insertions(+), 17 deletions(-)
> 
> diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
> index 78ea4d53..1f0c16b4 100644
> --- a/src/bin/proxmox-backup-proxy.rs
> +++ b/src/bin/proxmox-backup-proxy.rs
> @@ -1,4 +1,4 @@
> -use std::sync::{Arc};
> +use std::sync::{Arc, atomic::{AtomicUsize, Ordering}};
>   use std::path::{Path, PathBuf};
>   use std::os::unix::io::AsRawFd;
>   
> @@ -116,25 +116,12 @@ async fn run() -> Result<(), Error> {
>       let server = daemon::create_daemon(
>           ([0,0,0,0,0,0,0,0], 8007).into(),
>           |listener, ready| {
> -            let connections = proxmox_backup::tools::async_io::StaticIncoming::from(listener)
> -                .map_err(Error::from)
> -                .try_filter_map(move |(sock, _addr)| {
> -                    let acceptor = Arc::clone(&acceptor);
> -                    async move {
> -                        sock.set_nodelay(true).unwrap();
> -
> -                        let _ = set_tcp_keepalive(sock.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME);
>   
> -                        Ok(tokio_openssl::accept(&acceptor, sock)
> -                            .await
> -                            .ok() // handshake errors aren't be fatal, so return None to filter
> -                        )
> -                    }
> -                });
> -            let connections = proxmox_backup::tools::async_io::HyperAccept(connections);
> +            let connections = accept_connections(listener, acceptor);
> +            let connections = hyper::server::accept::from_stream(connections);
>   
>               Ok(ready
> -                .and_then(|_| hyper::Server::builder(connections)
> +               .and_then(|_| hyper::Server::builder(connections)
>                       .serve(rest_server)
>                       .with_graceful_shutdown(server::shutdown_future())
>                       .map_err(Error::from)
> @@ -170,6 +157,66 @@ async fn run() -> Result<(), Error> {
>       Ok(())
>   }
>   
> +fn accept_connections(
> +    mut listener: tokio::net::TcpListener,
> +    acceptor: Arc<openssl::ssl::SslAcceptor>,
> +) -> tokio::sync::mpsc::Receiver<Result<tokio_openssl::SslStream<tokio::net::TcpStream>, Error>> {
> +
> +    let (sender, receiver) = tokio::sync::mpsc::channel(100);
> +
> +    let accept_counter = Arc::new(AtomicUsize::new(0));
> +
> +    const MAX_PENDING_ACCEPTS: usize = 100;
> +
> +    tokio::spawn(async move {
> +        loop {
> +            match listener.accept().await {
> +                Err(err) => {
> +                    eprintln!("error accepting tcp connection: {}", err);
> +                }
> +                Ok((sock, _addr)) =>  {
> +                    sock.set_nodelay(true).unwrap();
> +                    let _ = set_tcp_keepalive(sock.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME);
> +                    let acceptor = Arc::clone(&acceptor);
> +                    let mut sender = sender.clone();
> +
> +                    if accept_counter.load(Ordering::SeqCst) > MAX_PENDING_ACCEPTS {
> +                        eprintln!("connection rejected - to many open connections");
> +                        continue;
> +                    }
> +                    accept_counter.fetch_add(1, Ordering::SeqCst);
> +
> +                    let accept_counter = accept_counter.clone();
> +                    tokio::spawn(async move {
> +                        let accept_future = tokio::time::timeout(
> +                            Duration::new(10, 0), tokio_openssl::accept(&acceptor, sock));
> +
> +                        let result = accept_future.await;
> +
> +                        match result {
> +                            Ok(Ok(connection)) => {
> +                                if let Err(_) = sender.send(Ok(connection)).await {
> +                                    eprintln!("detect closed connection channel");
> +                                }
> +                            }
> +                            Ok(Err(err)) => {
> +                                eprintln!("https handshakeX failed - {}", err);
> +                            }
> +                            Err(_) => {
> +                                eprintln!("https handshake timeout");
> +                            }
> +                        }
> +
> +                        accept_counter.fetch_sub(1, Ordering::SeqCst);
> +                    });
> +                }
> +            }
> +        }
> +    });
> +
> +    receiver
> +}
> +
>   fn start_stat_generator() {
>       let abort_future = server::shutdown_future();
>       let future = Box::pin(run_stat_generator());
>