Allow acquiring n permits from PollSemaphore

Adds `PollSemaphore::poll_acquire_many`, to allow acquiring `n` permits
from it.

tokio-util/src/sync/poll_semaphore.rs

@@ -39,7 +39,7 @@ impl PollSemaphore {
         self.semaphore
     }
 
-    /// Poll to acquire a permit from the semaphore.
+    /// Poll to acquire `n` permits from the semaphore.
     ///
     /// This can return the following values:
     ///
@@ -53,11 +53,28 @@ impl PollSemaphore {
     /// the `Waker` from the `Context` passed to the most recent call is
     /// scheduled to receive a wakeup.
     pub fn poll_acquire(&mut self, cx: &mut Context<'_>) -> Poll<Option<OwnedSemaphorePermit>> {
+        self.poll_acquire_many(1, cx)
+    }
+
+    /// Poll to acquire many permits from the semaphore.
+    ///
+    /// This can return the following values:
+    ///
+    ///  - `Poll::Pending` if a permit is not currently available.
+    ///  - `Poll::Ready(Some(permit))` if a permit was acquired.
+    ///  - `Poll::Ready(None)` if the semaphore has been closed.
+    ///
+    /// When this method returns `Poll::Pending`, the current task is scheduled
+    /// to receive a wakeup when the permits become available, or when the
+    /// semaphore is closed. Note that on multiple calls to `poll_acquire`, only
+    /// the `Waker` from the `Context` passed to the most recent call is
+    /// scheduled to receive a wakeup.
+    pub fn poll_acquire_many(&mut self, permits: u32, cx: &mut Context<'_>) -> Poll<Option<OwnedSemaphorePermit>> {
         let permit_future = match self.permit_fut.as_mut() {
             Some(fut) => fut,
             None => {
                 // avoid allocations completely if we can grab a permit immediately
-                match Arc::clone(&self.semaphore).try_acquire_owned() {
+                match Arc::clone(&self.semaphore).try_acquire_many_owned(permits) {
                     Ok(permit) => return Poll::Ready(Some(permit)),
                     Err(TryAcquireError::Closed) => return Poll::Ready(None),
                     Err(TryAcquireError::NoPermits) => {}