sync: consider exposing Semaphore::close in public API

[hawkw]

Is your feature request related to a problem? Please describe.
When updating tower::buffer to Tokio 0.3, it was necessary to replace the use of tokio's bounded MPSC channel with a re-implementation using the unbounded MPSC and Tokio's semaphore. This is because the bounded MPSC no longer exposes poll_ready, and the ready method borrows the sender, so it can't be easily boxed for an unbounded lifetime. Using Sender::poll_ready used to be sufficient to ensure that tasks waiting for the buffer service to become ready would be notified when the buffer worker terminates, because the senders would be woken on close.

This doesn't work with the new approach using the semaphore, because the Semaphore::acquire_owned method requires the Semaphore itself to be Arced. The Arc is necessary to uphold the intrusive waitlist's safety invariants; as the semaphore cannot be dropped while there are still tasks in its waitlist. However, this means we can't drop it to notify the waiting tasks. Instead, we use the somewhat jankier approach of just adding a big pile of permits to the semaphore and hoping that callers don't continue trying to acquire permits from it. This isn't great, for that reason and because it's theoretically (although probably not practically) possible to leak tasks if the semaphore is close to the maximum number of permits.

See tower-rs/tower#480 for details.

Describe the solution you'd like

The internal semaphore implementation has a close method, which does exactly what we'd want it to do:

tokio/tokio/src/sync/batch_semaphore.rs

Lines 165 to 186 in 9097ae5

	/// Closes the semaphore. This prevents the semaphore from issuing new
	/// permits and notifies all pending waiters.
	// This will be used once the bounded MPSC is updated to use the new
	// semaphore implementation.
	pub(crate) fn close(&self) {
	let mut waiters = self.waiters.lock();
	// If the semaphore's permits counter has enough permits for an
	// unqueued waiter to acquire all the permits it needs immediately,
	// it won't touch the wait list. Therefore, we have to set a bit on
	// the permit counter as well. However, we must do this while
	// holding the lock --- otherwise, if we set the bit and then wait
	// to acquire the lock we'll enter an inconsistent state where the
	// permit counter is closed, but the wait list is not.
	self.permits.fetch_or(Self::CLOSED, Release);
	waiters.closed = true;
	while let Some(mut waiter) = waiters.queue.pop_back() {
	let waker = unsafe { waiter.as_mut().waker.with_mut(|waker| (*waker).take()) };
	if let Some(waker) = waker {
	waker.wake();
	}
	}
	}

However, the public API semaphore type doesn't expose this for users to call. It would be nice if it did, for use cases like this.

Describe alternatives you've considered

• I also considered using a Notify alongside the semaphore to signal that the buffer has closed. However, Notify::notify_all is also not public, so that's not a great option.
• We could continue just adding a bunch of permits. This is probably fine but it feels kinda messy.

I can't think of any arguments against adding close to the API except that it's good to stabilize as minimal an API surface as possible. It doesn't feel like a potential footgun (as long as the behavior is documented)...

[carllerche]

This will require changing acquire() to return Result, making it a breaking change. I am tagging this for 1.0.

[carllerche]

Closed by #3065