sync: Add is_closed method to mpsc senders

[MikailBag]

Motivation

Closes #2469

Solution

I should note that I looked at mpsc internals first time, so I can be wrong / selected wrong implementation path.
I discovered that both Sender kinds (bounded and Unbounded) implement on top of tokio::sync::mpsc::Chan, which in turn uses Semaphore trait. I added new method is_closed to this trait. I was able to implement it properly for unbounded implementation.
However, I didn't find beautiful way to check if semaphore_ll is closed, so I implemented hack (we try to acquire new permit and immediately release it back). While it is inefficient and probably ugly, from the user side it should work OK (for example, it does not mess Sender internal state and it does not take mutable reference to sender).

[zeroed]

Thanks for exploring this issue and giving it a try; I'm still very interested in seeing an approach to this feature.
I did my exploration when it was marked as E-easy (^_^) but unfortunately I had to lose traction on it.

[LucioFranco]

@zeroed This was likely not a E-easy issue...sorry 😄

[LucioFranco]

@MikailBag looks like the loom ci test is failing, could you fix that up?

[LucioFranco]

Looks like a good and simple start, I want to make sure loom is happy though. Also would be good to add more tests.

[LucioFranco]

Would be good if we can open an issue for this TODO.

[MikailBag]

IIUC these are compile errors, because in loom mode Arc != std::sync::Arc.

I'll be able to fix these problems (and doc-comment) around 17th of August.

[LucioFranco]

@MikailBag 👍 no rush, feel free to ping me here or discord when you get a new revision up.

[zeroed]

@LucioFranco:

This was likely not a E-easy issue...sorry smile

Please, don't be: It has been an interesting learning experience. Mikail anticipated me with the "permits" approach and it's more than fine. Thanks!

[LucioFranco]

Do we need this implementation?

[MikailBag]

Do you mean unbounded channels should not have fn is_closed?

[hawkw]

I'm pretty sure that the semaphore's state field has a bit that we can read to check if it's closed:

tokio/tokio/src/sync/semaphore_ll.rs

Lines 116 to 117 in be7462e

	/// Semaphore has been closed, no more permits will be issued.
	const CLOSED: usize = 0b10;

For example, we use this here:

tokio/tokio/src/sync/semaphore_ll.rs

Lines 219 to 243 in be7462e

	// Load the current state
	let mut curr = SemState(self.state.load(Acquire));
	// Saves a ref to the waiter node
	let mut maybe_waiter: Option<NonNull<Waiter>> = None;
	/// Used in branches where we attempt to push the waiter into the wait
	/// queue but fail due to permits becoming available or the wait queue
	/// transitioning to "closed". In this case, the waiter must be
	/// transitioned back to the "idle" state.
	macro_rules! revert_to_idle {
	() => {
	if let Some(waiter) = maybe_waiter {
	unsafe { waiter.as_ref() }.revert_to_idle();
	}
	};
	}
	loop {
	let mut next = curr;
	if curr.is_closed() {
	revert_to_idle!();
	return Ready(Err(AcquireError::closed()));
	}

I think it should be possible to have an is_closed method on semaphore_ll that looks something like this:

impl Semaphore {
    // ...

    pub(crate) fn is_closed(&self) -> bool {
        SemState(self.state.load(Acquire)).is_closed()
    }
    
    // ...
}

There's additional logic that occurs when closing a semaphore, to assign permits to any pending waiters prior to marking the semaphore as closed. However, if this process is currently taking place, I think it's correct to not return true from is_closed yet. So AFAICT, this implementation ought to be valid.

[MikailBag]

I added is_closed method to semaphore_ll and now it is used instead of hacks with null waker.

[LucioFranco]

@hawkw can you take another look if you get the chance?

[MikailBag]

I've rebased onto master.
Now diff is pretty minimal.

[Darksonn]

I have some documentation suggestions.

[Darksonn]

This is my last comment.

[LucioFranco]

Thanks!