How to drop the environment of a closure passed to futures-cpupool?

Question

I have the following code:

extern crate futures;
extern crate futures_cpupool;
extern crate tokio_timer;

use std::time::Duration;

use futures::Future;
use futures_cpupool::CpuPool;
use tokio_timer::Timer;

fn work(foo: Foo) {
    std::thread::sleep(std::time::Duration::from_secs(10));
}

#[derive(Debug)]
struct Foo { }

impl Drop for Foo {
    fn drop(&mut self) {
        println!("Dropping Foo");
    }
}

fn main() {
    let pool = CpuPool::new_num_cpus();

    let foo = Foo { };

    let work_future = pool.spawn_fn(|| {
        let work = work(foo);

        let res: Result<(), ()> = Ok(work);
        res
    });

    println!("Created the future");

    let timer = Timer::default();
    let timeout = timer.sleep(Duration::from_millis(750))
        .then(|_| Err(()));

    let select = timeout.select(work_future).map(|(win, _)| win);

    match select.wait() {
        Ok(()) => { },
        Err(_) => { },
    }
}

It seems this code doesn't execute Foo::drop - no message is printed.

I expected foo to be dropped as soon as timeout future resolves in select, as it's a part of environment of a closure, passed to dropped future.

How to make it execute Foo::drop?

Answer 1

The documentation for CpuPool states:

The worker threads associated with a thread pool are kept alive so long as there is an open handle to the CpuPool or there is work running on them. Once all work has been drained and all references have gone away the worker threads will be shut down.

Additionally, you transfer ownership of foo from main to the closure, which then transfers it to work. work will drop foo at the end of the block. However, work is also performing a blocking sleep operation. This sleep counts as work running on the thread.

The sleep is still going when the main thread exits, which immediately tears down the program, and all the threads, without any time to clean up.

As pointed out in How to terminate or suspend a Rust thread from another thread? (and other questions in other languages), there's no safe way to terminate a thread.

I expected foo to be dropped as soon as timeout future resolves in select, as it's a part of environment of a closure, passed to dropped future.

The future doesn't actually "have" the closure or foo. All it has is a handle to the thread:

pub struct CpuFuture<T, E> {
    inner: Receiver<thread::Result<Result<T, E>>>,
    keep_running_flag: Arc<AtomicBool>,
}

Strangely, the docs say:

If the returned future is dropped then this CpuPool will attempt to cancel the computation, if possible. That is, if the computation is in the middle of working, it will be interrupted when possible.

However, I don't see any implementation for Drop for CpuFuture, so I don't see how it could be possible (or safe). Instead of Drop, the threadpool itself runs a Future. When that future is polled, it checks to see if the receiver has been dropped. This behavior is provided by the oneshot::Receiver. However, this has nothing to do with threads, which are outside the view of the future.