feat: make TerminatableTask terminate itself when dropped

[YuanYuYuan]

This PR introduces an auto termination when dropped and resolves the bug in #1054.

[YuanYuYuan]

@wyfo could you please review this? Thanks.

[wyfo]

I see why you changed the receiver type, but I think you should not try to reuse terminate in drop.
Instead, you can use some intermediate function:

async fn terminate_inner(handle: JoinHandle<()>, timeout: Duration) -> bool {
    if tokio::time::timeout(timeout, handle).await.is_err() {
        tracing::error!("Failed to terminate the task");
        return false;
    }
    true
}

And use it like

fn drop(&mut self) {
    if let Some(handle) = self.handle.take() {
        ResolveFuture::new(terminate_inner(handle, std::time::Duration::from_secs(10))).res_sync();
    }
}

fn terminate(mut self, timeout: Duration) -> bool {
    ResolveFuture::new(terminate_inner(self.handle.take().unwrap(), timeout)).res_sync();
}

fn terminate_async(mut self, timeout: Duration) -> bool {
    terminate_inner(self.handle.take().unwrap(), timeout).await
}

[wyfo]

I don't think you should change the receiver type of the method

[wyfo]

Idem about the receiver type

[YuanYuYuan]

If I'm not mistaken, to take out the value inside an Option requires &mut self. So we need to modify terminate and terminate_async in any way. Furthermore, we discussed with @Mallets today that moving out the ownership is not necessary as the borrow checker only lives in Rust, and mixing deconstruction and undeclaration could cause some conflicts. (You can check Undeckarable for more details.) We could follow the common pattern like https://docs.rs/tokio/latest/tokio/sync/mpsc/struct.Receiver.html#method.close. Closing a socket or receiver simply takes &mut self and hence Drop can coexist. I will create another draft PR on this topic further.

In conclusion, I suggest splitting the termination and the deconstruction of the task. @wyfo what do you think?

[wyfo]

If I'm not mistaken, to take out the value inside an Option requires &mut self. So we need to modify terminate and terminate_async in any way.

Just use mut self instead of self, and you can still use value semantic.

We could follow the common pattern like https://docs.rs/tokio/latest/tokio/sync/mpsc/struct.Receiver.html#method.close.

This is not the same thing here, as terminate returns a boolean. Do you want to store this boolean to return it on following calls?

Closing a socket or receiver simply takes &mut self and hence Drop can coexist.

There is no issue to coexist with drop and having terminate with self receiver.

[YuanYuYuan]

Hi @wyfo , thanks for you feedback.

Yes. That's what I mean we need to make some change in any way. Since I saw you wrote this in your first version, I thought you want to avoid any modification on self.

fn terminate(self, timeout: Duration) -> bool {
    ResolveFuture::new(terminate_inner(self.handle.take().unwrap(), timeout)).res_sync();
}

And I'm fine with the change of fn terminate(mut self).

The remaining issue is do we want to enforce a descrtuction here? This is related to the potential conflict with Drop. Let me explain it.

Originally, we define our TerminatableTask without Option

pub struct TerminatableTask {
    handle: JoinHandle<()>,
    token: CancellationToken,
}

This forbids us from implementing Drop for it. The conflict comes from

async fn terminate_inner(handle: JoinHandle<()>, timeout: Duration) -> bool {
    if tokio::time::timeout(timeout, handle).await.is_err() {
        tracing::error!("Failed to terminate the task");
        return false;
    }
    true
}
impl TerminatableTask {
    pub async fn terminate_async(self, timeout: Duration) -> bool {
        terminate_inner(self.handle, timeout).await  // ERROR: cannot move out of type `TerminatableTask`, which implements the `Drop` trait
    }
}

Passing self into the function potentially trigger a deconstruction or move out itself. There are same issues in the Undeclarable trait.
For instance, we implement Undeclarable trait for the Subscriber, then we can't implement Drop for it anymore.

impl<R> Drop for Subscriber<'_, R> {
    fn drop(&mut self) {
        todo!()
    }
}

(You can check this does cause lots of "cannot move out ..." errors)

So that's why I need to wrap the JoinHandle with a Option to not take out the handle but the inner value.

Back to this PR, either using &mut self or mut self can compile. But we still need to think twice: do we need to embed a deconstruction in terminate? As the example I pasted, closing a socket multiple times is acceptable in real use cases. In this PR, a 2nd termination on a closed task is simply doing nothing. This is exactly the same with the previous behavior in zenoh library (before the tokio porting).

Moreover, it can avoid any conflict with Drop if we implment/modify the TerminatableTask as below one day (which unfortunately has already existed in the case of Undeclarable).

impl TerminatableTask {
    pub async fn terminate_async(mut self, timeout: Duration) -> bool {
        let Self { token, mut handle } = self;  // ERROR: cannot move out of type `TerminatableTask`, which implements the `Drop` trait
        token.cancel();
        if let Some(handle) = handle.take() {
            if tokio::time::timeout(timeout, handle).await.is_err() {
                tracing::error!("Failed to terminate the task");
                return false;
            };
        }
        true
    }
}

BTW, I'm little bit confused with your point "should not try to reuse terminate in drop". TBH, I don't care about the return boolean especially it would report an error once it failed. And it seems that your example don't process the boolean result in Drop as well. Am I missing something?

[wyfo]

I give way to your arguments, it's good to me.