Convert async tokio timeout to sync

[modship]

For the conversion of async bootstrap to sync we need to convert this tokio timeout who is async.

tokio::time::timeout(timeout, future).await

I have tested a lot of stuff and here is a summary :

1. crossbeam channel with after()
2. thread + channel with drop(handle)
3. async_std cancel
4. future abort()
5. Unsafe kill thread

If you have any suggestion or know another way to do it i will be happy to test it.

---

1 : crossbeam channel with after() :

https://docs.rs/crossbeam-channel/latest/crossbeam_channel/fn.after.html#examples

fn main() {
    dbg!(crossbeam_select_timeout(Duration::from_secs(1)));
    println!("End of Main");
}

// crossbeam select version
fn crossbeam_select_timeout(timeout: Duration) -> Result<(), String> {
    let (tx, rx) = crossbeam::channel::bounded::<Result<(), String>>(1);

    std::thread::spawn(move || {
        // simulate some async stuff
        thread::sleep(Duration::from_secs(3));

        tx.send(Ok(())).unwrap();
    });

    select! {
        recv(after(timeout)) -> _ => Err("timeout".to_string()),
        recv(rx) -> _res => Ok(())
    }
}

[src/main.rs:12] crossbeam_select_timeout(Duration::from_secs(1)) = Err(
    "timeout",
)
End of Main

Sounds good but no really. if i create a loop in the spawned thread and make the main thread live :

fn main() {
    dbg!(crossbeam_select_timeout(Duration::from_secs(1)));
    println!("Continue the main thread for 5 sec");
    thread::sleep(Duration::from_secs(5));
    println!("End of Main");
}

// crossbeam select version
fn crossbeam_select_timeout(timeout: Duration) -> Result<(), String> {
    let (tx, rx) = crossbeam::channel::bounded::<Result<(), String>>(1);

     std::thread::spawn(move || {
        // simulate some async stuff
        loop {
            println!("future is alive");
            // simulate some async stuff that take time
            thread::sleep(Duration::from_secs(1));
        }
    });

    select! {
        recv(after(timeout)) -> _ => Err("timeout".to_string()),
        recv(rx) -> _res => Ok(())
    }
}

future is alive
future is alive
[src/main.rs:12] crossbeam_select_timeout(Duration::from_secs(1)) = Err(
    "timeout",
)
Continue the main thread for 5 sec
future is alive
future is alive
future is alive
future is alive
End of Main

The function is ended instantly when the timeout is reached but the async task continue to leave. In fact this is what we want when we spawn thread.
If we want to stop the spawned thread we want another solution.

---

2 : thread + channel with drop(handle)

fn main() {
    dbg!(timeout_thread(Duration::from_secs(1)));

    println!("Continue the main thread for 5 sec");
    thread::sleep(Duration::from_secs(5));
    println!("End of Main");
}

fn timeout_thread(timeout: Duration) -> Result<(), String> {
    let (tx, rx) = bounded(1);

    // timeout thread
    let tx2 = tx.clone();
    let fut = std::thread::spawn(move || {
        std::thread::sleep(timeout);
        tx2.send(Err("timeout".to_string())).unwrap();
    });

    let fut2 = std::thread::spawn(move || {
        // simulate some async stuff that take time
        loop {
            println!("fut2 alive");
            // simulate some async stuff that take time
            thread::sleep(Duration::from_secs(1));
        }
       // tx.send(Ok()).unwrap();
    });

    let res = rx.recv().unwrap();

    if res.is_ok() {
        // drop the thread timeout
        println!("cancel fut");
        drop(fut);
    } else {
        // here the second thread is dropped 
        println!("drop fut2");
        drop(fut2)
    }

    res
}

fut2 alive
fut2 alive
drop fut2
[src/main.rs:12] timeout_thread(Duration::from_secs(1)) = Err(
    "timeout",
)
Continue the main thread for 5 sec
fut2 alive
fut2 alive
fut2 alive
fut2 alive
End of Main

Same result as the first version

---

3 : async_std cancel

https://docs.rs/async-std/latest/async_std/task/struct.JoinHandle.html#method.cancel

fn main() {
    dbg!(async_std_cancel(Duration::from_secs(1)));

    println!("Continue the main thread for 5 sec");
    thread::sleep(Duration::from_secs(5));
    println!("End of Main");
}

fn async_std_cancel(timeout: Duration) -> Result<(), String> {
    let (tx, rx) = crossbeam::channel::bounded::<Result<(), String>>(1);
    let tx2 = tx.clone();
    let handle = async_std::task::spawn(async move {
        std::thread::sleep(timeout);
        tx2.send(Err("timeout".to_string())).unwrap();
    });

    let handle2 = async_std::task::spawn(async move {
        loop {
            println!("handle2 alive");
            // simulate some async stuff that take time
            thread::sleep(Duration::from_secs(1));
        }
     // tx.send(Ok(())).unwrap();
    });

    let res = rx.recv().unwrap();

    if res.is_ok() {
        // drop the thread timeout
        println!("Ok = cancel handle");
        async_std::task::block_on(handle.cancel());
    } else {
        // here the second thread is dropped because timeout
        println!("timeout = cancel handle2");
        let a = async_std::task::block_on(handle2.cancel());
    }

    res
}

handle2 alive
timeout = cancel handle2
handle2 alive
handle2 alive
handle2 alive
handle2 alive
handle2 alive
handle2 alive
....

Main thread is block by handle.cancel()

---

4 : future abort()

https://docs.rs/futures/latest/futures/future/struct.Abortable.html#method.new

fn main() {
    dbg!(future_abort(Duration::from_secs(1)));

    println!("Continue the main thread for 5 sec");
    thread::sleep(Duration::from_secs(5));
    println!("End of Main");
}

fn future_abort(timeout: Duration) -> Result<(), String> {
    let (abort_handle, abort_registration) = futures::future::AbortHandle::new_pair();
    let (tx, rx) = crossbeam::channel::bounded::<Result<(), String>>(1);

    // timeout thread
    let tx2 = tx.clone();
    let _fut_timeout = std::thread::spawn(move || {
        std::thread::sleep(timeout);
        tx2.send(Err("timeout".to_string())).unwrap();
    });

    let _future = Abortable::new(
        std::thread::spawn(move || {
            // simulate some async stuff that take time
            loop {
                println!("future is alive");
                thread::sleep(Duration::from_secs(1));
            }
        }),
        abort_registration,
    );

    let res = rx.recv().unwrap();

    if res.is_ok() {
        // todo abort the thread timeout
        println!("Ok = cancel timeout");
    } else {
        // here the second thread is abort
        println!("Err timeout = cancel future");
        abort_handle.abort();
    }

    res
}

future is alive
future is alive
Err timeout = cancel future
[src/main.rs:12] future_abort(Duration::from_secs(1)) = Err(
    "timeout",
)
Continue the main thread for 5 sec
future is alive
future is alive
future is alive
future is alive
End of Main

Same result as 1 && 2

---

5 : Unsafe kill thread

https://crates.io/crates/stop-thread
https://github.com/sollyucko/stop-thread

Their is one function for windows and one other for unix system.

fn main() {
    dbg!(kill_thread(Duration::from_secs(1)));

    println!("Continue the main thread for 5 sec");
    thread::sleep(Duration::from_secs(5));
    println!("End of Main");
}

fn kill_thread(timeout: Duration) -> Result<(), String> {
    let (tx, rx) = crossbeam::channel::bounded::<Result<(), String>>(1);

    // timeout thread
    let tx2 = tx.clone();
    let _fut_timeout = std::thread::spawn(move || {
        std::thread::sleep(timeout);
        tx2.send(Err("timeout".to_string())).unwrap();
    });

    let future = std::thread::spawn(move || {
        // simulate some async stuff that take time
        loop {
            println!("future is alive");
            // simulate some async stuff that take time
            thread::sleep(Duration::from_secs(1));
        }
    });

    let res = rx.recv().unwrap();

    if res.is_ok() {
        //  todo drop the thread timeout
        println!("Ok = kill timeout");
    } else {
        println!("Err timeout = kill the thread");
        // unix : stop_thread::kill_thread_graceful
        // windows : stop_thread::kill_thread_forcibly_exit_code
        unsafe {
            stop_thread::kill_thread_graceful(future);
        }
    }

    res
}

future is alive
future is alive
Err timeout = kill the thread
[src/main.rs:12] kill_thread(Duration::from_secs(1)) = Err(
    "timeout",
)
Continue the main thread for 5 sec
End of Main

With this unsafe crate the thread was instantly killed without waiting for the completion of the execution..

[Ben-PH]

#3633 is a checkpoint PR, and things seem to work well for non-timeout perpetual loops that get cancelled by a stop-signal. I believe mio has timeouts built into the system as well: might be interesting to investigate that?

As for option 5: that's a no-go for me. unsafe, and too much platform dependant behavior.

I think it's also important to clean up threads when needed, and not rely on the process stopping. I think mio works well with that: https://github.com/massalabs/massa/pull/3633/files#diff-2f2a0d279d4c015c632b27bdc7bd2fb62cf753fa8811dead19aadc3133666156R187-R193

[AurelienFT]

What's the use case of this timeout ?

For the writing and reading on network you can use the write_timeout and read_timeout of Rust TCP stream : https://doc.rust-lang.org/stable/std/net/struct.TcpStream.html#method.set_write_timeout

For global timeout any solution could work I personally like the crossbeam with timeout

[Ben-PH]

the write-timeout is good for individual writes, but write-all and read-all loops the individual writes. Each write is goverened by that timeout, but if it takes 100 writes to complete, you're 6-second timeout can take up to 10mins to trigger a timeout.

My checkpoint PR self review has a note in there already about how the io functions in the establisher.rs file "rub me the wrong way". That's in part coming from the question you raised.

[AurelienFT]

Ok will try to check the PR in the next days.

[Ben-PH]

The purpose was mostly to have a means to setup a to-do list of things to focus on with the diff. I probably should have forked and used a personal repo come to think of it.

By all means, have a look at it, just keep in mind that it's far from reviewable.