Auto merge of #124662 - zetanumbers:needs_async_drop, r=oli-obk

Implement `needs_async_drop` in rustc and optimize async drop glue

This PR expands on #121801 and implements `Ty::needs_async_drop` which works almost exactly the same as `Ty::needs_drop`, which is needed for #123948.

Also made compiler's async drop code to look more like compiler's regular drop code, which enabled me to write an optimization where types which do not use `AsyncDrop` can simply forward async drop glue to `drop_in_place`. This made size of the async block from the [async_drop test](https://github.com/zetanumbers/rust/blob/67980dd6fb11917d23d01a19c2cf4cfc3978aac8/tests/ui/async-await/async-drop.rs) to decrease by 12%.

16 files changed, 308 insertions, 161 deletions

diff --git a/compiler/rustc_hir/src/lang_items.rs b/compiler/rustc_hir/src/lang_items.rs
index f5f7bae11b2..c3ccba487ed 100644
--- a/compiler/rustc_hir/src/lang_items.rs
+++ b/compiler/rustc_hir/src/lang_items.rs
@@ -176,6 +176,7 @@ language_item_table! {
     AsyncDropSlice,          sym::async_drop_slice,    async_drop_slice_fn,        Target::Fn,             GenericRequirement::Exact(1);
     AsyncDropChain,          sym::async_drop_chain,    async_drop_chain_fn,        Target::Fn,             GenericRequirement::Exact(2);
     AsyncDropNoop,           sym::async_drop_noop,     async_drop_noop_fn,         Target::Fn,             GenericRequirement::Exact(0);
+    AsyncDropDeferredDropInPlace, sym::async_drop_deferred_drop_in_place, async_drop_deferred_drop_in_place_fn, Target::Fn, GenericRequirement::Exact(1);
     AsyncDropFuse,           sym::async_drop_fuse,     async_drop_fuse_fn,         Target::Fn,             GenericRequirement::Exact(1);
     AsyncDropDefer,          sym::async_drop_defer,    async_drop_defer_fn,        Target::Fn,             GenericRequirement::Exact(1);
     AsyncDropEither,         sym::async_drop_either,   async_drop_either_fn,       Target::Fn,             GenericRequirement::Exact(3);
diff --git a/compiler/rustc_hir_analysis/src/check/mod.rs b/compiler/rustc_hir_analysis/src/check/mod.rs
index 0083da2a1e4..149e7737e30 100644
--- a/compiler/rustc_hir_analysis/src/check/mod.rs
+++ b/compiler/rustc_hir_analysis/src/check/mod.rs
@@ -112,6 +112,7 @@ pub fn provide(providers: &mut Providers) {
     wfcheck::provide(providers);
     *providers = Providers {
         adt_destructor,
+        adt_async_destructor,
         region_scope_tree,
         collect_return_position_impl_trait_in_trait_tys,
         compare_impl_const: compare_impl_item::compare_impl_const_raw,
@@ -124,6 +125,10 @@ fn adt_destructor(tcx: TyCtxt<'_>, def_id: LocalDefId) -> Option<ty::Destructor>
     tcx.calculate_dtor(def_id.to_def_id(), dropck::check_drop_impl)
 }
 
+fn adt_async_destructor(tcx: TyCtxt<'_>, def_id: LocalDefId) -> Option<ty::AsyncDestructor> {
+    tcx.calculate_async_dtor(def_id.to_def_id(), dropck::check_drop_impl)
+}
+
 /// Given a `DefId` for an opaque type in return position, find its parent item's return
 /// expressions.
 fn get_owner_return_paths(
diff --git a/compiler/rustc_metadata/src/rmeta/decoder/cstore_impl.rs b/compiler/rustc_metadata/src/rmeta/decoder/cstore_impl.rs
index 4cad317ce80..f6b9c7ed992 100644
--- a/compiler/rustc_metadata/src/rmeta/decoder/cstore_impl.rs
+++ b/compiler/rustc_metadata/src/rmeta/decoder/cstore_impl.rs
@@ -287,6 +287,10 @@ provide! { tcx, def_id, other, cdata,
         let _ = cdata;
         tcx.calculate_dtor(def_id, |_,_| Ok(()))
     }
+    adt_async_destructor => {
+        let _ = cdata;
+        tcx.calculate_async_dtor(def_id, |_,_| Ok(()))
+    }
     associated_item_def_ids => {
         tcx.arena.alloc_from_iter(cdata.get_associated_item_or_field_def_ids(def_id.index))
     }
diff --git a/compiler/rustc_middle/src/query/erase.rs b/compiler/rustc_middle/src/query/erase.rs
index 1e36f034cc2..f98dbf8a0bd 100644
--- a/compiler/rustc_middle/src/query/erase.rs
+++ b/compiler/rustc_middle/src/query/erase.rs
@@ -238,6 +238,7 @@ trivial! {
     Option<rustc_hir::CoroutineKind>,
     Option<rustc_hir::HirId>,
     Option<rustc_middle::middle::stability::DeprecationEntry>,
+    Option<rustc_middle::ty::AsyncDestructor>,
     Option<rustc_middle::ty::Destructor>,
     Option<rustc_middle::ty::ImplTraitInTraitData>,
     Option<rustc_middle::ty::ScalarInt>,
@@ -295,6 +296,7 @@ trivial! {
     rustc_middle::ty::AssocItem,
     rustc_middle::ty::AssocItemContainer,
     rustc_middle::ty::Asyncness,
+    rustc_middle::ty::AsyncDestructor,
     rustc_middle::ty::BoundVariableKind,
     rustc_middle::ty::DeducedParamAttrs,
     rustc_middle::ty::Destructor,
diff --git a/compiler/rustc_middle/src/query/mod.rs b/compiler/rustc_middle/src/query/mod.rs
index 6ad4b7c40fb..d72ad09f954 100644
--- a/compiler/rustc_middle/src/query/mod.rs
+++ b/compiler/rustc_middle/src/query/mod.rs
@@ -703,6 +703,11 @@ rustc_queries! {
         cache_on_disk_if { key.is_local() }
         separate_provide_extern
     }
+    query adt_async_destructor(key: DefId) -> Option<ty::AsyncDestructor> {
+        desc { |tcx| "computing `AsyncDrop` impl for `{}`", tcx.def_path_str(key) }
+        cache_on_disk_if { key.is_local() }
+        separate_provide_extern
+    }
 
     query adt_sized_constraint(key: DefId) -> Option<ty::EarlyBinder<'tcx, Ty<'tcx>>> {
         desc { |tcx| "computing the `Sized` constraint for `{}`", tcx.def_path_str(key) }
@@ -1343,18 +1348,14 @@ rustc_queries! {
     query is_unpin_raw(env: ty::ParamEnvAnd<'tcx, Ty<'tcx>>) -> bool {
         desc { "computing whether `{}` is `Unpin`", env.value }
     }
-    /// Query backing `Ty::has_surface_async_drop`.
-    query has_surface_async_drop_raw(env: ty::ParamEnvAnd<'tcx, Ty<'tcx>>) -> bool {
-        desc { "computing whether `{}` has `AsyncDrop` implementation", env.value }
-    }
-    /// Query backing `Ty::has_surface_drop`.
-    query has_surface_drop_raw(env: ty::ParamEnvAnd<'tcx, Ty<'tcx>>) -> bool {
-        desc { "computing whether `{}` has `Drop` implementation", env.value }
-    }
     /// Query backing `Ty::needs_drop`.
     query needs_drop_raw(env: ty::ParamEnvAnd<'tcx, Ty<'tcx>>) -> bool {
         desc { "computing whether `{}` needs drop", env.value }
     }
+    /// Query backing `Ty::needs_async_drop`.
+    query needs_async_drop_raw(env: ty::ParamEnvAnd<'tcx, Ty<'tcx>>) -> bool {
+        desc { "computing whether `{}` needs async drop", env.value }
+    }
     /// Query backing `Ty::has_significant_drop_raw`.
     query has_significant_drop_raw(env: ty::ParamEnvAnd<'tcx, Ty<'tcx>>) -> bool {
         desc { "computing whether `{}` has a significant drop", env.value }
diff --git a/compiler/rustc_middle/src/ty/adt.rs b/compiler/rustc_middle/src/ty/adt.rs
index 8e946bc8b31..5f9b870331c 100644
--- a/compiler/rustc_middle/src/ty/adt.rs
+++ b/compiler/rustc_middle/src/ty/adt.rs
@@ -24,7 +24,9 @@ use std::hash::{Hash, Hasher};
 use std::ops::Range;
 use std::str;
 
-use super::{Destructor, FieldDef, GenericPredicates, Ty, TyCtxt, VariantDef, VariantDiscr};
+use super::{
+    AsyncDestructor, Destructor, FieldDef, GenericPredicates, Ty, TyCtxt, VariantDef, VariantDiscr,
+};
 
 #[derive(Clone, Copy, PartialEq, Eq, Hash, HashStable, TyEncodable, TyDecodable)]
 pub struct AdtFlags(u16);
@@ -577,6 +579,12 @@ impl<'tcx> AdtDef<'tcx> {
         tcx.adt_destructor(self.did())
     }
 
+    // FIXME: consider combining this method with `AdtDef::destructor` and removing
+    // this version
+    pub fn async_destructor(self, tcx: TyCtxt<'tcx>) -> Option<AsyncDestructor> {
+        tcx.adt_async_destructor(self.did())
+    }
+
     /// Returns a type such that `Self: Sized` if and only if that type is `Sized`,
     /// or `None` if the type is always sized.
     pub fn sized_constraint(self, tcx: TyCtxt<'tcx>) -> Option<ty::EarlyBinder<'tcx, Ty<'tcx>>> {
diff --git a/compiler/rustc_middle/src/ty/mod.rs b/compiler/rustc_middle/src/ty/mod.rs
index f24074cb472..90c154233da 100644
--- a/compiler/rustc_middle/src/ty/mod.rs
+++ b/compiler/rustc_middle/src/ty/mod.rs
@@ -1172,6 +1172,15 @@ pub struct Destructor {
     pub constness: hir::Constness,
 }
 
+// FIXME: consider combining this definition with regular `Destructor`
+#[derive(Copy, Clone, Debug, HashStable, Encodable, Decodable)]
+pub struct AsyncDestructor {
+    /// The `DefId` of the async destructor future constructor
+    pub ctor: DefId,
+    /// The `DefId` of the async destructor future type
+    pub future: DefId,
+}
+
 #[derive(Clone, Copy, PartialEq, Eq, HashStable, TyEncodable, TyDecodable)]
 pub struct VariantFlags(u8);
 bitflags::bitflags! {
diff --git a/compiler/rustc_middle/src/ty/sty.rs b/compiler/rustc_middle/src/ty/sty.rs
index 5cf96d29837..5f7385fccc9 100644
--- a/compiler/rustc_middle/src/ty/sty.rs
+++ b/compiler/rustc_middle/src/ty/sty.rs
@@ -25,7 +25,7 @@ use std::assert_matches::debug_assert_matches;
 use std::borrow::Cow;
 use std::iter;
 use std::ops::{ControlFlow, Range};
-use ty::util::IntTypeExt;
+use ty::util::{AsyncDropGlueMorphology, IntTypeExt};
 
 use rustc_type_ir::TyKind::*;
 use rustc_type_ir::{self as ir, BoundVar, CollectAndApply, DynKind};
@@ -1951,11 +1951,22 @@ impl<'tcx> Ty<'tcx> {
     }
 
     /// Returns the type of the async destructor of this type.
-    pub fn async_destructor_ty(self, tcx: TyCtxt<'tcx>, param_env: ParamEnv<'tcx>) -> Ty<'tcx> {
-        if self.is_async_destructor_noop(tcx, param_env) || matches!(self.kind(), ty::Error(_)) {
-            return Ty::async_destructor_combinator(tcx, LangItem::AsyncDropNoop)
-                .instantiate_identity();
+    pub fn async_destructor_ty(self, tcx: TyCtxt<'tcx>) -> Ty<'tcx> {
+        match self.async_drop_glue_morphology(tcx) {
+            AsyncDropGlueMorphology::Noop => {
+                return Ty::async_destructor_combinator(tcx, LangItem::AsyncDropNoop)
+                    .instantiate_identity();
+            }
+            AsyncDropGlueMorphology::DeferredDropInPlace => {
+                let drop_in_place =
+                    Ty::async_destructor_combinator(tcx, LangItem::AsyncDropDeferredDropInPlace)
+                        .instantiate(tcx, &[self.into()]);
+                return Ty::async_destructor_combinator(tcx, LangItem::AsyncDropFuse)
+                    .instantiate(tcx, &[drop_in_place.into()]);
+            }
+            AsyncDropGlueMorphology::Custom => (),
         }
+
         match *self.kind() {
             ty::Param(_) | ty::Alias(..) | ty::Infer(ty::TyVar(_)) => {
                 let assoc_items = tcx
@@ -1974,24 +1985,18 @@ impl<'tcx> Ty<'tcx> {
                 .adt_async_destructor_ty(
                     tcx,
                     adt_def.variants().iter().map(|v| v.fields.iter().map(|f| f.ty(tcx, args))),
-                    param_env,
                 ),
-            ty::Tuple(tys) => self.adt_async_destructor_ty(tcx, iter::once(tys), param_env),
-            ty::Closure(_, args) => self.adt_async_destructor_ty(
-                tcx,
-                iter::once(args.as_closure().upvar_tys()),
-                param_env,
-            ),
-            ty::CoroutineClosure(_, args) => self.adt_async_destructor_ty(
-                tcx,
-                iter::once(args.as_coroutine_closure().upvar_tys()),
-                param_env,
-            ),
+            ty::Tuple(tys) => self.adt_async_destructor_ty(tcx, iter::once(tys)),
+            ty::Closure(_, args) => {
+                self.adt_async_destructor_ty(tcx, iter::once(args.as_closure().upvar_tys()))
+            }
+            ty::CoroutineClosure(_, args) => self
+                .adt_async_destructor_ty(tcx, iter::once(args.as_coroutine_closure().upvar_tys())),
 
             ty::Adt(adt_def, _) => {
                 assert!(adt_def.is_union());
 
-                let surface_drop = self.surface_async_dropper_ty(tcx, param_env).unwrap();
+                let surface_drop = self.surface_async_dropper_ty(tcx).unwrap();
 
                 Ty::async_destructor_combinator(tcx, LangItem::AsyncDropFuse)
                     .instantiate(tcx, &[surface_drop.into()])
@@ -2008,17 +2013,12 @@ impl<'tcx> Ty<'tcx> {
         }
     }
 
-    fn adt_async_destructor_ty<I>(
-        self,
-        tcx: TyCtxt<'tcx>,
-        variants: I,
-        param_env: ParamEnv<'tcx>,
-    ) -> Ty<'tcx>
+    fn adt_async_destructor_ty<I>(self, tcx: TyCtxt<'tcx>, variants: I) -> Ty<'tcx>
     where
         I: Iterator + ExactSizeIterator,
         I::Item: IntoIterator<Item = Ty<'tcx>>,
     {
-        debug_assert!(!self.is_async_destructor_noop(tcx, param_env));
+        debug_assert_eq!(self.async_drop_glue_morphology(tcx), AsyncDropGlueMorphology::Custom);
 
         let defer = Ty::async_destructor_combinator(tcx, LangItem::AsyncDropDefer);
         let chain = Ty::async_destructor_combinator(tcx, LangItem::AsyncDropChain);
@@ -2041,7 +2041,7 @@ impl<'tcx> Ty<'tcx> {
             })
             .unwrap();
 
-        let dtor = if let Some(dropper_ty) = self.surface_async_dropper_ty(tcx, param_env) {
+        let dtor = if let Some(dropper_ty) = self.surface_async_dropper_ty(tcx) {
             Ty::async_destructor_combinator(tcx, LangItem::AsyncDropChain)
                 .instantiate(tcx, &[dropper_ty.into(), variants_dtor.into()])
         } else {
@@ -2052,21 +2052,13 @@ impl<'tcx> Ty<'tcx> {
             .instantiate(tcx, &[dtor.into()])
     }
 
-    fn surface_async_dropper_ty(
-        self,
-        tcx: TyCtxt<'tcx>,
-        param_env: ParamEnv<'tcx>,
-    ) -> Option<Ty<'tcx>> {
-        if self.has_surface_async_drop(tcx, param_env) {
-            Some(LangItem::SurfaceAsyncDropInPlace)
-        } else if self.has_surface_drop(tcx, param_env) {
-            Some(LangItem::AsyncDropSurfaceDropInPlace)
-        } else {
-            None
-        }
-        .map(|dropper| {
-            Ty::async_destructor_combinator(tcx, dropper).instantiate(tcx, &[self.into()])
-        })
+    fn surface_async_dropper_ty(self, tcx: TyCtxt<'tcx>) -> Option<Ty<'tcx>> {
+        let adt_def = self.ty_adt_def()?;
+        let dropper = adt_def
+            .async_destructor(tcx)
+            .map(|_| LangItem::SurfaceAsyncDropInPlace)
+            .or_else(|| adt_def.destructor(tcx).map(|_| LangItem::AsyncDropSurfaceDropInPlace))?;
+        Some(Ty::async_destructor_combinator(tcx, dropper).instantiate(tcx, &[self.into()]))
     }
 
     fn async_destructor_combinator(
diff --git a/compiler/rustc_middle/src/ty/util.rs b/compiler/rustc_middle/src/ty/util.rs
index cacaa859d52..52a0e72e17e 100644
--- a/compiler/rustc_middle/src/ty/util.rs
+++ b/compiler/rustc_middle/src/ty/util.rs
@@ -4,7 +4,7 @@ use crate::middle::codegen_fn_attrs::CodegenFnAttrFlags;
 use crate::query::{IntoQueryParam, Providers};
 use crate::ty::layout::{FloatExt, IntegerExt};
 use crate::ty::{
-    self, FallibleTypeFolder, Ty, TyCtxt, TypeFoldable, TypeFolder, TypeSuperFoldable,
+    self, Asyncness, FallibleTypeFolder, Ty, TyCtxt, TypeFoldable, TypeFolder, TypeSuperFoldable,
     TypeVisitableExt, Upcast,
 };
 use crate::ty::{GenericArgKind, GenericArgsRef};
@@ -382,6 +382,64 @@ impl<'tcx> TyCtxt<'tcx> {
         Some(ty::Destructor { did, constness })
     }
 
+    /// Calculate the async destructor of a given type.
+    pub fn calculate_async_dtor(
+        self,
+        adt_did: DefId,
+        validate: impl Fn(Self, DefId) -> Result<(), ErrorGuaranteed>,
+    ) -> Option<ty::AsyncDestructor> {
+        let async_drop_trait = self.lang_items().async_drop_trait()?;
+        self.ensure().coherent_trait(async_drop_trait).ok()?;
+
+        let ty = self.type_of(adt_did).instantiate_identity();
+        let mut dtor_candidate = None;
+        self.for_each_relevant_impl(async_drop_trait, ty, |impl_did| {
+            if validate(self, impl_did).is_err() {
+                // Already `ErrorGuaranteed`, no need to delay a span bug here.
+                return;
+            }
+
+            let [future, ctor] = self.associated_item_def_ids(impl_did) else {
+                self.dcx().span_delayed_bug(
+                    self.def_span(impl_did),
+                    "AsyncDrop impl without async_drop function or Dropper type",
+                );
+                return;
+            };
+
+            if let Some((_, _, old_impl_did)) = dtor_candidate {
+                self.dcx()
+                    .struct_span_err(self.def_span(impl_did), "multiple async drop impls found")
+                    .with_span_note(self.def_span(old_impl_did), "other impl here")
+                    .delay_as_bug();
+            }
+
+            dtor_candidate = Some((*future, *ctor, impl_did));
+        });
+
+        let (future, ctor, _) = dtor_candidate?;
+        Some(ty::AsyncDestructor { future, ctor })
+    }
+
+    /// Returns async drop glue morphology for a definition. To get async drop
+    /// glue morphology for a type see [`Ty::async_drop_glue_morphology`].
+    //
+    // FIXME: consider making this a query
+    pub fn async_drop_glue_morphology(self, did: DefId) -> AsyncDropGlueMorphology {
+        let ty: Ty<'tcx> = self.type_of(did).instantiate_identity();
+
+        // Async drop glue morphology is an internal detail, so reveal_all probably
+        // should be fine
+        let param_env = ty::ParamEnv::reveal_all();
+        if ty.needs_async_drop(self, param_env) {
+            AsyncDropGlueMorphology::Custom
+        } else if ty.needs_drop(self, param_env) {
+            AsyncDropGlueMorphology::DeferredDropInPlace
+        } else {
+            AsyncDropGlueMorphology::Noop
+        }
+    }
+
     /// Returns the set of types that are required to be alive in
     /// order to run the destructor of `def` (see RFCs 769 and
     /// 1238).
@@ -1138,6 +1196,18 @@ impl<'tcx> TypeFolder<TyCtxt<'tcx>> for WeakAliasTypeExpander<'tcx> {
     }
 }
 
+/// Indicates the form of `AsyncDestruct::Destructor`. Used to simplify async
+/// drop glue for types not using async drop.
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+pub enum AsyncDropGlueMorphology {
+    /// Async destructor simply does nothing
+    Noop,
+    /// Async destructor simply runs `drop_in_place`
+    DeferredDropInPlace,
+    /// Async destructor has custom logic
+    Custom,
+}
+
 impl<'tcx> Ty<'tcx> {
     /// Returns the `Size` for primitive types (bool, uint, int, char, float).
     pub fn primitive_size(self, tcx: TyCtxt<'tcx>) -> Size {
@@ -1303,80 +1373,16 @@ impl<'tcx> Ty<'tcx> {
         }
     }
 
-    /// Checks whether values of this type `T` implements the `AsyncDrop`
-    /// trait.
-    pub fn has_surface_async_drop(self, tcx: TyCtxt<'tcx>, param_env: ty::ParamEnv<'tcx>) -> bool {
-        self.could_have_surface_async_drop() && tcx.has_surface_async_drop_raw(param_env.and(self))
-    }
-
-    /// Fast path helper for testing if a type has `AsyncDrop`
-    /// implementation.
-    ///
-    /// Returning `false` means the type is known to not have `AsyncDrop`
-    /// implementation. Returning `true` means nothing -- could be
-    /// `AsyncDrop`, might not be.
-    fn could_have_surface_async_drop(self) -> bool {
-        !self.is_async_destructor_trivially_noop()
-            && !matches!(
-                self.kind(),
-                ty::Tuple(_)
-                    | ty::Slice(_)
-                    | ty::Array(_, _)
-                    | ty::Closure(..)
-                    | ty::CoroutineClosure(..)
-                    | ty::Coroutine(..)
-            )
-    }
-
-    /// Checks whether values of this type `T` implements the `Drop`
-    /// trait.
-    pub fn has_surface_drop(self, tcx: TyCtxt<'tcx>, param_env: ty::ParamEnv<'tcx>) -> bool {
-        self.could_have_surface_drop() && tcx.has_surface_drop_raw(param_env.and(self))
-    }
-
-    /// Fast path helper for testing if a type has `Drop` implementation.
-    ///
-    /// Returning `false` means the type is known to not have `Drop`
-    /// implementation. Returning `true` means nothing -- could be
-    /// `Drop`, might not be.
-    fn could_have_surface_drop(self) -> bool {
-        !self.is_async_destructor_trivially_noop()
-            && !matches!(
-                self.kind(),
-                ty::Tuple(_)
-                    | ty::Slice(_)
-                    | ty::Array(_, _)
-                    | ty::Closure(..)
-                    | ty::CoroutineClosure(..)
-                    | ty::Coroutine(..)
-            )
-    }
-
-    /// Checks whether values of this type `T` implement has noop async destructor.
+    /// Get morphology of the async drop glue, needed for types which do not
+    /// use async drop. To get async drop glue morphology for a definition see
+    /// [`TyCtxt::async_drop_glue_morphology`]. Used for `AsyncDestruct::Destructor`
+    /// type construction.
     //
-    // FIXME: implement optimization to make ADTs, which do not need drop,
-    // to skip fields or to have noop async destructor.
-    pub fn is_async_destructor_noop(
-        self,
-        tcx: TyCtxt<'tcx>,
-        param_env: ty::ParamEnv<'tcx>,
-    ) -> bool {
-        self.is_async_destructor_trivially_noop()
-            || if let ty::Adt(adt_def, _) = self.kind() {
-                (adt_def.is_union() || adt_def.is_payloadfree())
-                    && !self.has_surface_async_drop(tcx, param_env)
-                    && !self.has_surface_drop(tcx, param_env)
-            } else {
-                false
-            }
-    }
-
-    /// Fast path helper for testing if a type has noop async destructor.
-    ///
-    /// Returning `true` means the type is known to have noop async destructor
-    /// implementation. Returning `true` means nothing -- could be
-    /// `Drop`, might not be.
-    fn is_async_destructor_trivially_noop(self) -> bool {
+    // FIXME: implement optimization to not instantiate a certain morphology of
+    // async drop glue too soon to allow per type optimizations, see array case
+    // for more info. Perhaps then remove this method and use `needs_(async_)drop`
+    // instead.
+    pub fn async_drop_glue_morphology(self, tcx: TyCtxt<'tcx>) -> AsyncDropGlueMorphology {
         match self.kind() {
             ty::Int(_)
             | ty::Uint(_)
@@ -1388,10 +1394,43 @@ impl<'tcx> Ty<'tcx> {
             | ty::Ref(..)
             | ty::RawPtr(..)
             | ty::FnDef(..)
-            | ty::FnPtr(_) => true,
-            ty::Tuple(tys) => tys.is_empty(),
-            ty::Adt(adt_def, _) => adt_def.is_manually_drop(),
-            _ => false,
+            | ty::FnPtr(_)
+            | ty::Infer(ty::FreshIntTy(_))
+            | ty::Infer(ty::FreshFloatTy(_)) => AsyncDropGlueMorphology::Noop,
+
+            ty::Tuple(tys) if tys.is_empty() => AsyncDropGlueMorphology::Noop,
+            ty::Adt(adt_def, _) if adt_def.is_manually_drop() => AsyncDropGlueMorphology::Noop,
+
+            // Foreign types can never have destructors.
+            ty::Foreign(_) => AsyncDropGlueMorphology::Noop,
+
+            // FIXME: implement dynamic types async drops
+            ty::Error(_) | ty::Dynamic(..) => AsyncDropGlueMorphology::DeferredDropInPlace,
+
+            ty::Tuple(_) | ty::Array(_, _) | ty::Slice(_) => {
+                // Assume worst-case scenario, because we can instantiate async
+                // destructors in different orders:
+                //
+                // 1. Instantiate [T; N] with T = String and N = 0
+                // 2. Instantiate <[String; 0] as AsyncDestruct>::Destructor
+                //
+                // And viceversa, thus we cannot rely on String not using async
+                // drop or array having zero (0) elements
+                AsyncDropGlueMorphology::Custom
+            }
+            ty::Pat(ty, _) => ty.async_drop_glue_morphology(tcx),
+
+            ty::Adt(adt_def, _) => tcx.async_drop_glue_morphology(adt_def.did()),
+
+            ty::Closure(did, _)
+            | ty::CoroutineClosure(did, _)
+            | ty::Coroutine(did, _)
+            | ty::CoroutineWitness(did, _) => tcx.async_drop_glue_morphology(*did),
+
+            ty::Alias(..) | ty::Param(_) | ty::Bound(..) | ty::Placeholder(..) | ty::Infer(_) => {
+                // No specifics, but would usually mean forwarding async drop glue
+                AsyncDropGlueMorphology::Custom
+            }
         }
     }
 
@@ -1430,6 +1469,46 @@ impl<'tcx> Ty<'tcx> {
         }
     }
 
+    /// If `ty.needs_async_drop(...)` returns `true`, then `ty` is definitely
+    /// non-copy and *might* have a async destructor attached; if it returns
+    /// `false`, then `ty` definitely has no async destructor (i.e., no async
+    /// drop glue).
+    ///
+    /// (Note that this implies that if `ty` has an async destructor attached,
+    /// then `needs_async_drop` will definitely return `true` for `ty`.)
+    ///
+    /// When constructing `AsyncDestruct::Destructor` type, use
+    /// [`Ty::async_drop_glue_morphology`] instead.
+    //
+    // FIXME(zetanumbers): Note that this method is used to check eligible types
+    // in unions.
+    #[inline]
+    pub fn needs_async_drop(self, tcx: TyCtxt<'tcx>, param_env: ty::ParamEnv<'tcx>) -> bool {
+        // Avoid querying in simple cases.
+        match needs_drop_components(tcx, self) {
+            Err(AlwaysRequiresDrop) => true,
+            Ok(components) => {
+                let query_ty = match *components {
+                    [] => return false,
+                    // If we've got a single component, call the query with that
+                    // to increase the chance that we hit the query cache.
+                    [component_ty] => component_ty,
+                    _ => self,
+                };
+
+                // This doesn't depend on regions, so try to minimize distinct
+                // query keys used.
+                // If normalization fails, we just use `query_ty`.
+                debug_assert!(!param_env.has_infer());
+                let query_ty = tcx
+                    .try_normalize_erasing_regions(param_env, query_ty)
+                    .unwrap_or_else(|_| tcx.erase_regions(query_ty));
+
+                tcx.needs_async_drop_raw(param_env.and(query_ty))
+            }
+        }
+    }
+
     /// Checks if `ty` has a significant drop.
     ///
     /// Note that this method can return false even if `ty` has a destructor
@@ -1601,10 +1680,25 @@ impl<'tcx> ExplicitSelf<'tcx> {
 /// Returns a list of types such that the given type needs drop if and only if
 /// *any* of the returned types need drop. Returns `Err(AlwaysRequiresDrop)` if
 /// this type always needs drop.
+//
+// FIXME(zetanumbers): consider replacing this with only
+// `needs_drop_components_with_async`
+#[inline]
 pub fn needs_drop_components<'tcx>(
     tcx: TyCtxt<'tcx>,
     ty: Ty<'tcx>,
 ) -> Result<SmallVec<[Ty<'tcx>; 2]>, AlwaysRequiresDrop> {
+    needs_drop_components_with_async(tcx, ty, Asyncness::No)
+}
+
+/// Returns a list of types such that the given type needs drop if and only if
+/// *any* of the returned types need drop. Returns `Err(AlwaysRequiresDrop)` if
+/// this type always needs drop.
+pub fn needs_drop_components_with_async<'tcx>(
+    tcx: TyCtxt<'tcx>,
+    ty: Ty<'tcx>,
+    asyncness: Asyncness,
+) -> Result<SmallVec<[Ty<'tcx>; 2]>, AlwaysRequiresDrop> {
     match *ty.kind() {
         ty::Infer(ty::FreshIntTy(_))
         | ty::Infer(ty::FreshFloatTy(_))
@@ -1623,11 +1717,18 @@ pub fn needs_drop_components<'tcx>(
         // Foreign types can never have destructors.
         ty::Foreign(..) => Ok(SmallVec::new()),
 
-        ty::Dynamic(..) | ty::Error(_) => Err(AlwaysRequiresDrop),
+        // FIXME(zetanumbers): Temporary workaround for async drop of dynamic types
+        ty::Dynamic(..) | ty::Error(_) => {
+            if asyncness.is_async() {
+                Ok(SmallVec::new())
+            } else {
+                Err(AlwaysRequiresDrop)
+            }
+        }
 
-        ty::Pat(ty, _) | ty::Slice(ty) => needs_drop_components(tcx, ty),
+        ty::Pat(ty, _) | ty::Slice(ty) => needs_drop_components_with_async(tcx, ty, asyncness),
         ty::Array(elem_ty, size) => {
-            match needs_drop_components(tcx, elem_ty) {
+            match needs_drop_components_with_async(tcx, elem_ty, asyncness) {
                 Ok(v) if v.is_empty() => Ok(v),
                 res => match size.try_to_target_usize(tcx) {
                     // Arrays of size zero don't need drop, even if their element
@@ -1643,7 +1744,7 @@ pub fn needs_drop_components<'tcx>(
         }
         // If any field needs drop, then the whole tuple does.
         ty::Tuple(fields) => fields.iter().try_fold(SmallVec::new(), move |mut acc, elem| {
-            acc.extend(needs_drop_components(tcx, elem)?);
+            acc.extend(needs_drop_components_with_async(tcx, elem, asyncness)?);
             Ok(acc)
         }),
 
diff --git a/compiler/rustc_mir_transform/src/shim/async_destructor_ctor.rs b/compiler/rustc_mir_transform/src/shim/async_destructor_ctor.rs
index f4481c22fc1..aa9c87d8f80 100644
--- a/compiler/rustc_mir_transform/src/shim/async_destructor_ctor.rs
+++ b/compiler/rustc_mir_transform/src/shim/async_destructor_ctor.rs
@@ -12,7 +12,7 @@ use rustc_middle::mir::{
     Terminator, TerminatorKind, UnwindAction, UnwindTerminateReason, RETURN_PLACE,
 };
 use rustc_middle::ty::adjustment::PointerCoercion;
-use rustc_middle::ty::util::Discr;
+use rustc_middle::ty::util::{AsyncDropGlueMorphology, Discr};
 use rustc_middle::ty::{self, Ty, TyCtxt};
 use rustc_middle::{bug, span_bug};
 use rustc_span::source_map::respan;
@@ -116,15 +116,25 @@ impl<'tcx> AsyncDestructorCtorShimBuilder<'tcx> {
     }
 
     fn build(self) -> Body<'tcx> {
-        let (tcx, def_id, Some(self_ty)) = (self.tcx, self.def_id, self.self_ty) else {
+        let (tcx, Some(self_ty)) = (self.tcx, self.self_ty) else {
             return self.build_zst_output();
         };
+        match self_ty.async_drop_glue_morphology(tcx) {
+            AsyncDropGlueMorphology::Noop => span_bug!(
+                self.span,
+                "async drop glue shim generator encountered type with noop async drop glue morphology"
+            ),
+            AsyncDropGlueMorphology::DeferredDropInPlace => {
+                return self.build_deferred_drop_in_place();
+            }
+            AsyncDropGlueMorphology::Custom => (),
+        }
 
         let surface_drop_kind = || {
-            let param_env = tcx.param_env_reveal_all_normalized(def_id);
-            if self_ty.has_surface_async_drop(tcx, param_env) {
+            let adt_def = self_ty.ty_adt_def()?;
+            if adt_def.async_destructor(tcx).is_some() {
                 Some(SurfaceDropKind::Async)
-            } else if self_ty.has_surface_drop(tcx, param_env) {
+            } else if adt_def.destructor(tcx).is_some() {
                 Some(SurfaceDropKind::Sync)
             } else {
                 None
@@ -267,6 +277,13 @@ impl<'tcx> AsyncDestructorCtorShimBuilder<'tcx> {
         self.return_()
     }
 
+    fn build_deferred_drop_in_place(mut self) -> Body<'tcx> {
+        self.put_self();
+        let deferred = self.combine_deferred_drop_in_place();
+        self.combine_fuse(deferred);
+        self.return_()
+    }
+
     fn build_fused_async_surface(mut self) -> Body<'tcx> {
         self.put_self();
         let surface = self.combine_async_surface();
@@ -441,6 +458,14 @@ impl<'tcx> AsyncDestructorCtorShimBuilder<'tcx> {
         )
     }
 
+    fn combine_deferred_drop_in_place(&mut self) -> Ty<'tcx> {
+        self.apply_combinator(
+            1,
+            LangItem::AsyncDropDeferredDropInPlace,
+            &[self.self_ty.unwrap().into()],
+        )
+    }
+
     fn combine_fuse(&mut self, inner_future_ty: Ty<'tcx>) -> Ty<'tcx> {
         self.apply_combinator(1, LangItem::AsyncDropFuse, &[inner_future_ty.into()])
     }
@@ -481,7 +506,7 @@ impl<'tcx> AsyncDestructorCtorShimBuilder<'tcx> {
         if let Some(ty) = self.self_ty {
             debug_assert_eq!(
                 output.ty(&self.locals, self.tcx),
-                ty.async_destructor_ty(self.tcx, self.param_env),
+                ty.async_destructor_ty(self.tcx),
                 "output async destructor types did not match for type: {ty:?}",
             );
         }
diff --git a/compiler/rustc_span/src/symbol.rs b/compiler/rustc_span/src/symbol.rs
index 360e355416c..61ca0d54ca4 100644
--- a/compiler/rustc_span/src/symbol.rs
+++ b/compiler/rustc_span/src/symbol.rs
@@ -429,6 +429,7 @@ symbols! {
         async_drop,
         async_drop_chain,
         async_drop_defer,
+        async_drop_deferred_drop_in_place,
         async_drop_either,
         async_drop_fuse,
         async_drop_in_place,
diff --git a/compiler/rustc_trait_selection/src/solve/normalizes_to/mod.rs b/compiler/rustc_trait_selection/src/solve/normalizes_to/mod.rs
index f9e164e1f3f..22a77e2ecb8 100644
--- a/compiler/rustc_trait_selection/src/solve/normalizes_to/mod.rs
+++ b/compiler/rustc_trait_selection/src/solve/normalizes_to/mod.rs
@@ -845,7 +845,7 @@ impl<'tcx> assembly::GoalKind<'tcx> for NormalizesTo<'tcx> {
             | ty::Str
             | ty::Slice(_)
             | ty::Tuple(_)
-            | ty::Error(_) => self_ty.async_destructor_ty(ecx.interner(), goal.param_env),
+            | ty::Error(_) => self_ty.async_destructor_ty(ecx.interner()),
 
             // We do not call `Ty::async_destructor_ty` on alias, param, or placeholder
             // types, which return `<self_ty as AsyncDestruct>::AsyncDestructor`
diff --git a/compiler/rustc_trait_selection/src/traits/project.rs b/compiler/rustc_trait_selection/src/traits/project.rs
index b6557e6c4eb..501d9c9266e 100644
--- a/compiler/rustc_trait_selection/src/traits/project.rs
+++ b/compiler/rustc_trait_selection/src/traits/project.rs
@@ -1560,7 +1560,7 @@ fn confirm_builtin_candidate<'cx, 'tcx>(
         let destructor_def_id = tcx.associated_item_def_ids(trait_def_id)[0];
         assert_eq!(destructor_def_id, item_def_id);
 
-        (self_ty.async_destructor_ty(tcx, obligation.param_env).into(), Vec::new())
+        (self_ty.async_destructor_ty(tcx).into(), Vec::new())
     } else if lang_items.pointee_trait() == Some(trait_def_id) {
         let metadata_def_id = tcx.require_lang_item(LangItem::Metadata, None);
         assert_eq!(metadata_def_id, item_def_id);
diff --git a/compiler/rustc_ty_utils/src/common_traits.rs b/compiler/rustc_ty_utils/src/common_traits.rs
index cb95239e991..51b908881eb 100644
--- a/compiler/rustc_ty_utils/src/common_traits.rs
+++ b/compiler/rustc_ty_utils/src/common_traits.rs
@@ -22,17 +22,6 @@ fn is_unpin_raw<'tcx>(tcx: TyCtxt<'tcx>, query: ty::ParamEnvAnd<'tcx, Ty<'tcx>>)
     is_item_raw(tcx, query, LangItem::Unpin)
 }
 
-fn has_surface_async_drop_raw<'tcx>(
-    tcx: TyCtxt<'tcx>,
-    query: ty::ParamEnvAnd<'tcx, Ty<'tcx>>,
-) -> bool {
-    is_item_raw(tcx, query, LangItem::AsyncDrop)
-}
-
-fn has_surface_drop_raw<'tcx>(tcx: TyCtxt<'tcx>, query: ty::ParamEnvAnd<'tcx, Ty<'tcx>>) -> bool {
-    is_item_raw(tcx, query, LangItem::Drop)
-}
-
 fn is_item_raw<'tcx>(
     tcx: TyCtxt<'tcx>,
     query: ty::ParamEnvAnd<'tcx, Ty<'tcx>>,
@@ -45,13 +34,5 @@ fn is_item_raw<'tcx>(
 }
 
 pub(crate) fn provide(providers: &mut Providers) {
-    *providers = Providers {
-        is_copy_raw,
-        is_sized_raw,
-        is_freeze_raw,
-        is_unpin_raw,
-        has_surface_async_drop_raw,
-        has_surface_drop_raw,
-        ..*providers
-    };
+    *providers = Providers { is_copy_raw, is_sized_raw, is_freeze_raw, is_unpin_raw, ..*providers };
 }
diff --git a/compiler/rustc_ty_utils/src/instance.rs b/compiler/rustc_ty_utils/src/instance.rs
index 0d089205c1e..e4dcea785d4 100644
--- a/compiler/rustc_ty_utils/src/instance.rs
+++ b/compiler/rustc_ty_utils/src/instance.rs
@@ -4,6 +4,7 @@ use rustc_infer::infer::TyCtxtInferExt;
 use rustc_middle::bug;
 use rustc_middle::query::Providers;
 use rustc_middle::traits::{BuiltinImplSource, CodegenObligationError};
+use rustc_middle::ty::util::AsyncDropGlueMorphology;
 use rustc_middle::ty::GenericArgsRef;
 use rustc_middle::ty::{self, Instance, TyCtxt, TypeVisitableExt};
 use rustc_span::sym;
@@ -59,7 +60,7 @@ fn resolve_instance<'tcx>(
         } else if Some(def_id) == tcx.lang_items().async_drop_in_place_fn() {
             let ty = args.type_at(0);
 
-            if !ty.is_async_destructor_noop(tcx, param_env) {
+            if ty.async_drop_glue_morphology(tcx) != AsyncDropGlueMorphology::Noop {
                 match *ty.kind() {
                     ty::Closure(..)
                     | ty::CoroutineClosure(..)
diff --git a/compiler/rustc_ty_utils/src/needs_drop.rs b/compiler/rustc_ty_utils/src/needs_drop.rs
index 72ee1a48249..205b3f2760f 100644
--- a/compiler/rustc_ty_utils/src/needs_drop.rs
+++ b/compiler/rustc_ty_utils/src/needs_drop.rs
@@ -30,6 +30,21 @@ fn needs_drop_raw<'tcx>(tcx: TyCtxt<'tcx>, query: ty::ParamEnvAnd<'tcx, Ty<'tcx>
     res
 }
 
+fn needs_async_drop_raw<'tcx>(tcx: TyCtxt<'tcx>, query: ty::ParamEnvAnd<'tcx, Ty<'tcx>>) -> bool {
+    // If we don't know a type doesn't need async drop, for example if it's a
+    // type parameter without a `Copy` bound, then we conservatively return that
+    // it needs async drop.
+    let adt_has_async_dtor =
+        |adt_def: ty::AdtDef<'tcx>| adt_def.async_destructor(tcx).map(|_| DtorType::Significant);
+    let res = drop_tys_helper(tcx, query.value, query.param_env, adt_has_async_dtor, false)
+        .filter(filter_array_elements(tcx, query.param_env))
+        .next()
+        .is_some();
+
+    debug!("needs_drop_raw({:?}) = {:?}", query, res);
+    res
+}
+
 /// HACK: in order to not mistakenly assume that `[PhantomData<T>; N]` requires drop glue
 /// we check the element type for drop glue. The correct fix would be looking at the
 /// entirety of the code around `needs_drop_components` and this file and come up with
@@ -389,6 +404,7 @@ fn adt_significant_drop_tys(
 pub(crate) fn provide(providers: &mut Providers) {
     *providers = Providers {
         needs_drop_raw,
+        needs_async_drop_raw,
         has_significant_drop_raw,
         adt_drop_tys,
         adt_significant_drop_tys,