TcpListener::accept() in tokio::spawn() freeze the app

[xolatgames]

Hi folks! Here's me again .

async fn start_ssh_tunnel(password: String) -> Result<(), Box<dyn std::error::Error>> {
    let addr = format!("{}:{}", SSH_HOST, SSH_PORT);

    let ssh_client = IHandler {};
    let mut session = russh::client::connect(
        Arc::new(Config::default()),
        addr,
        ssh_client
    ).await?;
    session.authenticate_password(SSH_USER, password).await?;

    let listener = TcpListener::bind(SocketAddrV4::new(Ipv4Addr::LOCALHOST, FORWARDING_PORT)).await?;
    let addr = listener.local_addr()?;

    tokio::spawn(async move {
        let (mut socket, _) = listener.accept().await.unwrap();

        let channel = session.channel_open_direct_tcpip(
            SSH_HOST, FORWARDING_PORT as u32,
            Ipv4Addr::LOCALHOST.to_string(), addr.port() as u32
        ).await.unwrap();
        let mut stream = channel.into_stream();

        tokio::io::copy_bidirectional(
            &mut socket, &mut stream
        ).await.unwrap();
    });

    Ok(())
}

I want to implement an SSH forwarding. Some problems begins from here:

let (mut socket, _) = listener.accept().await.unwrap();

The app just hangs in this moment, and doesn't want to execute next commands.

I've check it by adding something like this:

...

println!("Hello!");

let (mut socket, _) = listener.accept().await.unwrap();

println!("Hello!");

let channel = session.channel_open_direct_tcpip(
    SSH_HOST, FORWARDING_PORT as u32,
    Ipv4Addr::LOCALHOST.to_string(), addr.port() as u32
).await.unwrap();

...

Hello! text prints only once in a terminal.

The function above calls from this funtion:

...

pub fn connect(password: String) -> Result<Backend, String> {
    let runtime = tokio::runtime::Runtime::new().map_err(|e| format!("Failed to create async runtime: {e}"))?;

    runtime.block_on(async { 
        start_ssh_tunnel(password).await
    }).map_err(|e| format!("Failed to create SSH tunnel: {e}"))?;
    
    ...
}

...

That is not async.

Please, let me know - how to fix that?

[nerditation]

what do you mean by "just hang"?

the tcp port is open and waiting for connections, did you try to connect to the port but couldn't? is there a firewall rule to block the port?

[xolatgames]

Hello!

I mean that the app stucks. A window with the app shows "Not responding" inside itself (it's the feature of GNOME in Linux), with a few options ("Wait", and "Stop"), and I can't do something anymore, only stop it. But before - I was can.

About firewall - yes, it's running, and I forgot to set it up. I'll try to set it up, then return to here.

[xolatgames]

English - isn't my native language, so I don't know some words in this .

[xolatgames]

I forgot that I've set up firewall for blocking incoming packets, but don't for outgoing. Here's the hot, so my brains won't works . But I've add rules for allowing incoming packets by the necessarily ports, and obviously, it still won't works .

[SkiFire13]

@xolatgames

I mean that the app stucks. A window with the app shows "Not responding" inside itself

There's no code that handles window in the snippet you posted. Did you forget something?

[xolatgames]

Hello!

It's possible. I'll try to send you more parts of the code now...

[xolatgames]

It's main.rs:

...

use interface::Interface;
use eframe::egui;

fn main() -> eframe::Result<()> {
	const APP_NAME: &str = "Some app";

    const WIN_WIDTH: f32 = 640.;
    const WIN_HEIGHT: f32 = 480.;

    let options = eframe::NativeOptions {
        viewport: egui::ViewportBuilder::default()
            .with_inner_size([WIN_WIDTH, WIN_HEIGHT])
            .with_min_inner_size([WIN_WIDTH, WIN_HEIGHT])
            .with_title(APP_NAME),
        ..Default::default()
    };

    eframe::run_native(
        APP_NAME,
        options,
        Box::new(|_cc| Box::new(Interface::default()))
    )
}

It's interface.rs:

use eframe::egui;
use crate::pages::{
    connection::Connection,
    app::App,
};

pub enum Pages {
    CONNECTION,
    APP,
}

pub struct Interface {
    current_page: Pages,
    p_connection: Connection,
    p_app: App,
}

impl Default for Interface {
    fn default() -> Self {
        let p_connection = Connection::default();
        let p_app = App::default();

        Self {
            current_page: Pages::CONNECTION,
            p_connection,
            p_app,
        }
    }
}

impl Interface {
    fn press_connection(&mut self) {
        match self.p_app.connect(self.p_connection.get_password()) {
            Ok(_) => {
                self.current_page = Pages::APP;
            },
            Err(err) => {
                self.p_connection.set_error(err);
            },
        }
    }
}

impl eframe::Interface for Interface {
    fn update(&mut self, ctx: &egui::Context, _frame: &mut eframe::Frame) {
        match self.current_page {
            Pages::CONNECTION => {
                if self.p_connection.connect_pressed {
                    self.press_connection();
                    self.p_connection.connect_pressed = false;
                }

                self.p_connection.update(ctx, _frame);
            },
            Pages::APP => {
                self.p_app.update(ctx, _frame);
            },
        }
    }
}

The app isn't going to App page, it stucks at Connection page.

It's app.rs:

...

use crate::backend::AppBackend;

impl App {
    pub fn connect(&mut self, password: String) {
        let (backend, backend_error) = match AppBackend::connect(password) {
            Ok(backend) => (Some(backend), None),
            Err(err) => (None, Some(err)),
        };

        ...

        self.backend = backend;
        
        ...
    }
    
    ...
}

...

But it's App's backend:

...

pub struct AppBackend {
    runtime: tokio::runtime::Runtime,

    ...
}

...

impl AppBackend {
    pub fn connect(password: String) -> Result<AppBackend, String> {
        let runtime = tokio::runtime::Runtime::new().map_err(|e| format!("Failed to create async runtime: {e}"))?;

        runtime.block_on(async { 
            start_ssh_tunnel(password).await
        }).map_err(|e| format!("Failed to create SSH tunnel: {e}"))?;

        ...

        Ok(Self {
            runtime,

            ...
        })
    }

    ...
}

...

And it's exactly the block of the code where's the problem:

...

async fn start_ssh_tunnel(password: String) -> Result<(), Box<dyn std::error::Error>> {
    let addr = format!("{}:{}", SSH_HOST, SSH_PORT);

    let ssh_client = IHandler {};
    let mut session = russh::client::connect(
        Arc::new(Config::default()),
        addr,
        ssh_client
    ).await?;
    session.authenticate_password(SSH_USER, password).await?;

    let listener = TcpListener::bind(SocketAddrV4::new(Ipv4Addr::LOCALHOST, FORWARDING_PORT)).await?;
    let addr = listener.local_addr()?;

    tokio::spawn(async move {
        let (mut socket, _) = listener.accept().await.unwrap();

        let channel = session.channel_open_direct_tcpip(
            SSH_HOST, FORWARDING_PORT as u32,
            Ipv4Addr::LOCALHOST.to_string(), addr.port() as u32
        ).await.unwrap();
        let mut stream = channel.into_stream();

        tokio::io::copy_bidirectional(
            &mut socket, &mut stream
        ).await.unwrap();
    });

    Ok(())
}

Here's the problem:

let (mut socket, _) = listener.accept().await.unwrap();

The app stucks exactly here .

[SkiFire13]

You end up calling runtime.block_on from the UI thread, so your UI will completely freeze until your future completes.

[xolatgames]

How can I fix that?

[xolatgames]

Freeze... Forgot about this word in English... Thanks .

[nerditation]

@xolatgames

    pub fn connect(password: String) -> Result<AppBackend, String> {
        let runtime = tokio::runtime::Runtime::new().map_err(|e| format!("Failed to create async runtime: {e}"))?;

        runtime.block_on(async { 
            start_ssh_tunnel(password).await
        }).map_err(|e| format!("Failed to create SSH tunnel: {e}"))?;

you are creating a tokio runtime inside this connect() function, and block_on() will wait for the async task to complete, and blocks the event loop of the gui.

block_on() is not what you want to set the runtime context, use enter() instead.

remember, you cannot block the ui thread, always use spawn() for async tasks. if you need to display the result of the async task in the gui, use a channel to send the data back, and call request_repaint() in the async task to wakeup the ui thread.

also, don't create a temporary runtime inside the function, move it to main(). (in theory, you can manage the runtime as part of the application state, but the enter guard is a borrowed form, so you'll need someway to workaround the self-referential problem.)

// in main, before entering the gui event loop, enter a tokio runtime first:
fn main() {
    //...
    let rt = Runtime::new().unwrap();
    let _guard = rt.enter();
    eframe::run_native(...)
}

then you need a WaitForConnection transient page, something like this;

match self.current_page {
    Pages::CONNECTION => {
        // if the "connect" button is pressed, start the async task and move to the "wait" page
        if connection_pressed {
            let egui_ctx = ctx.clone();
            let (tx, rx) = tokio::oneshot::channel();
            self.connected_notifier = Some(rx);
            tokio::spawn(async move {
                let result = Backend::connect(password).await;
                // send the async result back to the ui
                tx.send(result);
                // wakeup the ui thread
                ctx.request_repaint();
            });
        }
        self.current_page = PAGES::WAIT_FOR_CONNECTION;
    }
    Pages::WAIT_FOR_CONNECTION => {
        // ui might be wakeup by normal events, use `try_recv()` to check
        if let Ok(connect_result) = self.connected_notifier.as_mut().try_recv() {
            // the async task finished, check the result and move to next page
            todo!("process async result");
            self.current_page = Pages::APP;
        }
        Pages::APP => {
            todo!("normal app logic");
        }
    }
}

if you don't like this manual approach, try egui-async, which abstracted away the low level details. there are several examples to get you started quickly.

GitHub - xangelix/egui-async: A simple library for running async tasks in egui...

[xolatgames]

Thanks too much!

I'll try it. But firstly I need to go to eat.

[xolatgames]

Hello everyone! I wish ya a good day!

@nerditation, I've move tokio's runtime to the main function, and have try to pass it to the Interface's function that creates an Interface's structure for next working with it:

fn main() -> eframe::Result<(), String> {
    ...

    let runtime = Runtime::new()
        .map_err(|e| format!("Failed to create async runtime: {e}"))?;
    let _guard = runtime.enter();

    ...
    
    eframe::run_native(
        APP_NAME,
        options,
        Box::new(|_cc| Box::new(Interface::new(&runtime)))
    )
        .map_err(|e| format!("Error in interface's GUI loop: {e}"))?;

    Ok(())
}

But I've got the error from this idea:

error[E0597]: `runtime` does not live long enough
  --> src/main.rs:33:43
   |
18 |     let runtime = Runtime::new()
   |         ------- binding `runtime` declared here
...
33 |         Box::new(|_cc| Box::new(Interface::new(&runtime)))
   |         ----------------------------------^^^^^^^---
   |         |        |                        |
   |         |        |                        borrowed value does not live long enough
   |         |        value captured here
   |         coercion requires that `runtime` is borrowed for `'static`
...
38 | }
   | - `runtime` dropped here while still borrowed

This is the function that creates Interface's structure:

impl Interface<'_> {
    pub fn new<'a>(runtime: &'a Runtime) -> Interface<'a> {
        let p_connection = Connection::default();
        let p_app = App::new(runtime);

        Interface {
            runtime: Some(runtime),

            current_page: Pages::CONNECTION,
            p_connection,
            p_app,
        }
    }

    ...
}

Also, the compiler, as the rust-analyzer, can't find tokio::oneshot::channel();.

How to fix those issues?

Thanks to anyone in advance!

[nerditation]

@xolatgames

impl Interface<'_> {
    pub fn new<'a>(runtime: &'a Runtime) -> Interface<'a> {
        let p_connection = Connection::default();
        let p_app = App::new(runtime);
...

you don't need to store a reference of the runtime in your app state, just leave it in main. when you call Runtime::enter(), it will be become the async "context" for the calling thread, and then everything just works.

I don't know why you want to store the runtime, but there's nothing you can do with a &Runtime that you cannot do with a Handle. you can get the Handle within an async context using Handle::current().

@xolatgames

can't find tokio::oneshot::channel();

it's actually tokio::sync::oneshot::channel();, you need to enable the sync feature.

channel in tokio::sync::oneshot - Rust

[xolatgames]

Thanks too much . I'll try to figured out with it a little bit later, then let you know if there's something is will going wrong.