refactor(ext/http): use scopeguard defer to handle async drop (#20652)

Use the [scopeguard](https://docs.rs/scopeguard/) defer macro to run
cleanup code for `new_slab_future`.
This means it can be a single async function, avoiding the need to
create a struct and implement `PinnedDrop`

Async cleanup in Rust is awkward because async functions may be
cancelled at any await point when their Future is dropped.
The scopeguard approach comes from the following articles:
* [How to think about `async`/`await` in
Rust](http://cliffle.com/blog/async-inversion/)
* [Async Cancellation
I](https://blog.yoshuawuyts.com/async-cancellation-1/) (Reddit
[discussion](https://www.reddit.com/r/rust/comments/qrhg39/blog_post_async_cancellation/))

1 files changed, 22 insertions, 0 deletions

diff --git a/ext/http/slab.rs b/ext/http/slab.rs
index 7c48b87b6..790b4649a 100644
--- a/ext/http/slab.rs
+++ b/ext/http/slab.rs
@@ -10,6 +10,7 @@ use http::HeaderMap;
 use hyper1::body::Incoming;
 use hyper1::upgrade::OnUpgrade;
 
+use scopeguard::defer;
 use slab::Slab;
 use std::cell::RefCell;
 use std::cell::RefMut;
@@ -52,6 +53,27 @@ impl Drop for HttpRequestBodyAutocloser {
   }
 }
 
+pub async fn new_slab_future(
+  request: Request,
+  request_info: HttpConnectionProperties,
+  refcount: RefCount,
+  tx: tokio::sync::mpsc::Sender<SlabId>,
+) -> Result<Response, hyper::Error> {
+  let index = slab_insert(request, request_info, refcount);
+  defer! {
+    slab_drop(index);
+  }
+  let rx = slab_get(index).promise();
+  if tx.send(index).await.is_ok() {
+    http_trace!(index, "SlabFuture await");
+    // We only need to wait for completion if we aren't closed
+    rx.await;
+    http_trace!(index, "SlabFuture complete");
+  }
+  let response = slab_get(index).take_response();
+  Ok(response)
+}
+
 pub struct HttpSlabRecord {
   request_info: HttpConnectionProperties,
   request_parts: Parts,