Repository: bytecodealliance/wasmtime
Pull request #8611 (2024-05-14 00:09:20)

wasi: Handle read(0) with file streams
The `read` method on streams should support a size argument of zero, to test if the stream is still open. However, the implementation of `read` for open files was unconditionally treating a read that would return zero bytes as an indication that the stream was closed.

This PR fixes this bug by handling a `read(0)` as an explicit test to see if the current position is at the end of the file.

<!--
Please make sure you include the following information:

- If this work has been discussed elsewhere, please include a link to that
  conversation. If it was discussed in an issue, just mention "issue #...".

- Explain why this change is needed. If the details are in an issue already,
  this can be brief.

Our development process is documented in the Wasmtime book:
https://docs.wasmtime.dev/contributing-development-process.html

Please ensure all communication follows the code of conduct:
https://github.com/bytecodealliance/wasmtime/blob/main/CODE_OF_CONDUCT.md
-->



Patch:
diff --git a/Cargo.toml b/Cargo.toml
index 80e362307..9efec7e6b 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -263,7 +263,7 @@ async-trait = "0.1.71"
 heck = "0.4"
 similar = "2.1.0"
 toml = "0.8.10"
-mach2 = "0.4.1"
+mach2 = "0.4.2"
 memfd = "0.6.2"
 memoffset = "0.9.0"
 psm = "0.1.11"
diff --git a/benches/wasmtime-serve-rps.sh b/benches/wasmtime-serve-rps.sh
index 19262b539..33d83a74a 100755
--- a/benches/wasmtime-serve-rps.sh
+++ b/benches/wasmtime-serve-rps.sh
@@ -15,12 +15,16 @@ set -e
 
 repo_dir="$(dirname $0)/.."
 cargo_toml="$repo_dir/Cargo.toml"
+target_dir="$CARGO_TARGET_DIR"
+if [[ "$target_dir" == "" ]]; then
+    target_dir="$repo_dir/target"
+fi
 
 # Build Wasmtime.
 cargo build --manifest-path "$cargo_toml" --release -p wasmtime-cli
 
 # Spawn `wasmtime serve` in the background.
-cargo run --manifest-path "$cargo_toml" --release -- serve "$@" &
+"$target_dir/release/wasmtime" serve "$@" &
 pid=$!
 
 # Give it a second to print its diagnostic information and get the server up and
diff --git a/crates/cli-flags/src/lib.rs b/crates/cli-flags/src/lib.rs
index 50ddc9441..fce7d87e1 100644
--- a/crates/cli-flags/src/lib.rs
+++ b/crates/cli-flags/src/lib.rs
@@ -63,9 +63,19 @@ wasmtime_option_group! {
         /// linear memories.
         pub guard_before_linear_memory: Option<bool>,
 
+        /// Whether to initialize tables lazily, so that instantiation is
+        /// fast but indirect calls are a little slower. If no, tables are
+        /// initialized eagerly from any active element segments that apply to
+        /// them during instantiation. (default: yes)
+        pub table_lazy_init: Option<bool>,
+
         /// Enable the pooling allocator, in place of the on-demand allocator.
         pub pooling_allocator: Option<bool>,
 
+        /// The number of decommits to do per batch. A batch size of 1
+        /// effectively disables decommit batching. (default: 1)
+        pub pooling_decommit_batch_size: Option<u32>,
+
         /// How many bytes to keep resident between instantiations for the
         /// pooling allocator in linear memories.
         pub pooling_memory_keep_resident: Option<usize>,
@@ -544,6 +554,9 @@ impl CommonOptions {
         if let Some(enable) = self.opts.guard_before_linear_memory {
             config.guard_before_linear_memory(enable);
         }
+        if let Some(enable) = self.opts.table_lazy_init {
+            config.table_lazy_init(enable);
+        }
 
         // If fuel has been configured, set the `consume fuel` flag on the config.
         if self.wasm.fuel.is_some() {
diff --git a/crates/cranelift/src/func_environ.rs b/crates/cranelift/src/func_environ.rs
index b8319cbfe..869a9874c 100644
--- a/crates/cranelift/src/func_environ.rs
+++ b/crates/cranelift/src/func_environ.rs
@@ -817,6 +817,7 @@ impl<'module_environment> FuncEnvironment<'module_environment> {
         table_index: TableIndex,
         index: ir::Value,
         cold_blocks: bool,
+        lazy_init: bool,
     ) -> ir::Value {
         let pointer_type = self.pointer_type();
         self.ensure_table_exists(builder.func, table_index);
@@ -833,6 +834,11 @@ impl<'module_environment> FuncEnvironment<'module_environment> {
             self.isa.flags().enable_table_access_spectre_mitigation(),
         );
         let value = builder.ins().load(pointer_type, flags, table_entry_addr, 0);
+
+        if !lazy_init {
+            return value;
+        }
+
         // Mask off the "initialized bit". See documentation on
         // FUNCREF_INIT_BIT in crates/environ/src/ref_bits.rs for more
         // details. Note that `FUNCREF_MASK` has type `usize` which may not be
@@ -1354,11 +1360,14 @@ impl<'a, 'func, 'module_env> Call<'a, 'func, 'module_env> {
         cold_blocks: bool,
     ) -> WasmResult<Option<(ir::Value, ir::Value)>> {
         // Get the funcref pointer from the table.
+        let table = &self.env.module.table_plans[table_index];
+        let TableStyle::CallerChecksSignature { lazy_init } = table.style;
         let funcref_ptr = self.env.get_or_init_func_ref_table_elem(
             self.builder,
             table_index,
             callee,
             cold_blocks,
+            lazy_init,
         );
 
         // If necessary, check the signature.
@@ -1407,7 +1416,7 @@ impl<'a, 'func, 'module_env> Call<'a, 'func, 'module_env> {
 
         // Generate a rustc compile error here if more styles are added in
         // the future as the following code is tailored to just this style.
-        let TableStyle::CallerChecksSignature = table.style;
+        let TableStyle::CallerChecksSignature { .. } = table.style;
 
         // Test if a type check is necessary for this table. If this table is a
         // table of typed functions and that type matches `ty_index`, then
@@ -1752,9 +1761,14 @@ impl<'module_environment> cranelift_wasm::FuncEnvironment for FuncEnvironment<'m
 
             // Function types.
             WasmHeapTopType::Func => match plan.style {
-                TableStyle::CallerChecksSignature => {
-                    Ok(self.get_or_init_func_ref_table_elem(builder, table_index, index, false))
-                }
+                TableStyle::CallerChecksSignature { lazy_init } => Ok(self
+                    .get_or_init_func_ref_table_elem(
+                        builder,
+                        table_index,
+                        index,
+                        false,
+                        lazy_init,
+                    )),
             },
         }
     }
@@ -1805,7 +1819,7 @@ impl<'module_environment> cranelift_wasm::FuncEnvironment for FuncEnvironment<'m
             // Function types.
             WasmHeapTopType::Func => {
                 match plan.style {
-                    TableStyle::CallerChecksSignature => {
+                    TableStyle::CallerChecksSignature { lazy_init } => {
                         let (elem_addr, flags) = table_data.prepare_table_addr(
                             builder,
                             index,
@@ -1815,9 +1829,13 @@ impl<'module_environment> cranelift_wasm::FuncEnvironment for FuncEnvironment<'m
                         // Set the "initialized bit". See doc-comment on
                         // `FUNCREF_INIT_BIT` in
                         // crates/environ/src/ref_bits.rs for details.
-                        let value_with_init_bit = builder
-                            .ins()
-                            .bor_imm(value, Imm64::from(FUNCREF_INIT_BIT as i64));
+                        let value_with_init_bit = if lazy_init {
+                            builder
+                                .ins()
+                                .bor_imm(value, Imm64::from(FUNCREF_INIT_BIT as i64))
+                        } else {
+                            value
+                        };
                         builder
                             .ins()
                             .store(flags, value_with_init_bit, elem_addr, 0);
diff --git a/crates/environ/src/module.rs b/crates/environ/src/module.rs
index 9e25280ef..6e505b4f3 100644
--- a/crates/environ/src/module.rs
+++ b/crates/environ/src/module.rs
@@ -315,13 +315,19 @@ pub trait InitMemory {
 #[derive(Debug, Clone, Hash, Serialize, Deserialize)]
 pub enum TableStyle {
     /// Signatures are stored in the table and checked in the caller.
-    CallerChecksSignature,
+    CallerChecksSignature {
+        /// Whether this table is initialized lazily and requires an
+        /// initialization check on every access.
+        lazy_init: bool,
+    },
 }
 
 impl TableStyle {
     /// Decide on an implementation style for the given `Table`.
-    pub fn for_table(_table: Table, _tunables: &Tunables) -> Self {
-        Self::CallerChecksSignature
+    pub fn for_table(_table: Table, tunables: &Tunables) -> Self {
+        Self::CallerChecksSignature {
+            lazy_init: tunables.table_lazy_init,
+        }
     }
 }
 
diff --git a/crates/environ/src/tunables.rs b/crates/environ/src/tunables.rs
index 440c593e9..f945b858b 100644
--- a/crates/environ/src/tunables.rs
+++ b/crates/environ/src/tunables.rs
@@ -42,6 +42,12 @@ pub struct Tunables {
     /// beginning of the allocation in addition to the end.
     pub guard_before_linear_memory: bool,
 
+    /// Whether to initialize tables lazily, so that instantiation is fast but
+    /// indirect calls are a little slower. If false, tables are initialized
+    /// eagerly from any active element segments that apply to them during
+    /// instantiation.
+    pub table_lazy_init: bool,
+
     /// Indicates whether an address map from compiled native code back to wasm
     /// offsets in the original file is generated.
     pub generate_address_map: bool,
@@ -113,6 +119,7 @@ impl Tunables {
             epoch_interruption: false,
             static_memory_bound_is_maximum: false,
             guard_before_linear_memory: true,
+            table_lazy_init: true,
             generate_address_map: true,
             debug_adapter_modules: false,
             relaxed_simd_deterministic: false,
diff --git a/crates/fiber/src/lib.rs b/crates/fiber/src/lib.rs
index 6292079e3..b889a7c38 100644
--- a/crates/fiber/src/lib.rs
+++ b/crates/fiber/src/lib.rs
@@ -61,6 +61,12 @@ impl FiberStack {
     pub fn range(&self) -> Option<Range<usize>> {
         self.0.range()
     }
+
+    /// Is this a manually-managed stack created from raw parts? If so, it is up
+    /// to whoever created it to manage the stack's memory allocation.
+    pub fn is_from_raw_parts(&self) -> bool {
+        self.0.is_from_raw_parts()
+    }
 }
 
 /// A creator of RuntimeFiberStacks.
@@ -82,7 +88,7 @@ pub unsafe trait RuntimeFiberStack: Send + Sync {
 }
 
 pub struct Fiber<'a, Resume, Yield, Return> {
-    stack: FiberStack,
+    stack: Option<FiberStack>,
     inner: imp::Fiber,
     done: Cell<bool>,
     _phantom: PhantomData<&'a (Resume, Yield, Return)>,
@@ -114,7 +120,7 @@ impl<'a, Resume, Yield, Return> Fiber<'a, Resume, Yield, Return> {
         let inner = imp::Fiber::new(&stack.0, func)?;
 
         Ok(Self {
-            stack,
+            stack: Some(stack),
             inner,
             done: Cell::new(false),
             _phantom: PhantomData,
@@ -139,7 +145,7 @@ impl<'a, Resume, Yield, Return> Fiber<'a, Resume, Yield, Return> {
     pub fn resume(&self, val: Resume) -> Result<Return, Yield> {
         assert!(!self.done.replace(true), "cannot resume a finished fiber");
         let result = Cell::new(RunResult::Resuming(val));
-        self.inner.resume(&self.stack.0, &result);
+        self.inner.resume(&self.stack().0, &result);
         match result.into_inner() {
             RunResult::Resuming(_) | RunResult::Executing => unreachable!(),
             RunResult::Yield(y) => {
@@ -158,7 +164,13 @@ impl<'a, Resume, Yield, Return> Fiber<'a, Resume, Yield, Return> {
 
     /// Gets the stack associated with this fiber.
     pub fn stack(&self) -> &FiberStack {
-        &self.stack
+        self.stack.as_ref().unwrap()
+    }
+
+    /// When this fiber has finished executing, reclaim its stack.
+    pub fn into_stack(mut self) -> FiberStack {
+        assert!(self.done());
+        self.stack.take().unwrap()
     }
 }
 
diff --git a/crates/fiber/src/unix.rs b/crates/fiber/src/unix.rs
index f14a5cf53..b16006543 100644
--- a/crates/fiber/src/unix.rs
+++ b/crates/fiber/src/unix.rs
@@ -41,7 +41,7 @@ pub struct FiberStack {
 
     /// Stored here to ensure that when this `FiberStack` the backing storage,
     /// if any, is additionally dropped.
-    _storage: FiberStackStorage,
+    storage: FiberStackStorage,
 }
 
 enum FiberStackStorage {
@@ -66,7 +66,7 @@ impl FiberStack {
         Ok(FiberStack {
             base: stack.mapping_base.wrapping_byte_add(page_size),
             len: stack.mapping_len - page_size,
-            _storage: FiberStackStorage::Mmap(stack),
+            storage: FiberStackStorage::Mmap(stack),
         })
     }
 
@@ -79,10 +79,14 @@ impl FiberStack {
         Ok(FiberStack {
             base,
             len,
-            _storage: FiberStackStorage::Unmanaged,
+            storage: FiberStackStorage::Unmanaged,
         })
     }
 
+    pub fn is_from_raw_parts(&self) -> bool {
+        matches!(self.storage, FiberStackStorage::Unmanaged)
+    }
+
     pub fn from_custom(custom: Box<dyn RuntimeFiberStack>) -> io::Result<Self> {
         let range = custom.range();
         let page_size = rustix::param::page_size();
@@ -99,7 +103,7 @@ impl FiberStack {
         Ok(FiberStack {
             base: start_ptr,
             len: range.len(),
-            _storage: FiberStackStorage::Custom(custom),
+            storage: FiberStackStorage::Custom(custom),
         })
     }
 
diff --git a/crates/fiber/src/windows.rs b/crates/fiber/src/windows.rs
index 9a0b3497e..e15d1bb2b 100644
--- a/crates/fiber/src/windows.rs
+++ b/crates/fiber/src/windows.rs
@@ -19,6 +19,10 @@ impl FiberStack {
         Err(io::Error::from_raw_os_error(ERROR_NOT_SUPPORTED as i32))
     }
 
+    pub fn is_from_raw_parts(&self) -> bool {
+        false
+    }
+
     pub fn from_custom(_custom: Box<dyn RuntimeFiberStack>) -> io::Result<Self> {
         Err(io::Error::from_raw_os_error(ERROR_NOT_SUPPORTED as i32))
     }
diff --git a/crates/fuzzing/src/generators/config.rs b/crates/fuzzing/src/generators/config.rs
index 13ba248a0..b77dfa026 100644
--- a/crates/fuzzing/src/generators/config.rs
+++ b/crates/fuzzing/src/generators/config.rs
@@ -227,6 +227,9 @@ impl Config {
             }
 
             cfg.cranelift_pcc(pcc);
+
+            // Eager init is currently only supported on Cranelift, not Winch.
+            cfg.table_lazy_init(self.wasmtime.table_lazy_init);
         }
 
         self.wasmtime.async_config.configure(&mut cfg);
@@ -483,6 +486,7 @@ pub struct WasmtimeConfig {
     cache_call_indirects: bool,
     /// The maximum number of call-indirect cache slots.
     max_call_indirect_cache_slots: usize,
+    table_lazy_init: bool,
 
     /// Whether or not fuzzing should enable PCC.
     pcc: bool,
diff --git a/crates/fuzzing/src/generators/pooling_config.rs b/crates/fuzzing/src/generators/pooling_config.rs
index 0997597f0..32ecc8c4d 100644
--- a/crates/fuzzing/src/generators/pooling_config.rs
+++ b/crates/fuzzing/src/generators/pooling_config.rs
@@ -27,6 +27,7 @@ pub struct PoolingAllocationConfig {
     pub table_keep_resident: usize,
     pub linear_memory_keep_resident: usize,
 
+    pub decommit_batch_size: usize,
     pub max_unused_warm_slots: u32,
 
     pub async_stack_zeroing: bool,
@@ -61,6 +62,7 @@ impl PoolingAllocationConfig {
         cfg.table_keep_resident(self.table_keep_resident);
         cfg.linear_memory_keep_resident(self.linear_memory_keep_resident);
 
+        cfg.decommit_batch_size(self.decommit_batch_size);
         cfg.max_unused_warm_slots(self.max_unused_warm_slots);
 
         cfg.async_stack_zeroing(self.async_stack_zeroing);
@@ -106,6 +108,7 @@ impl<'a> Arbitrary<'a> for PoolingAllocationConfig {
             table_keep_resident: u.int_in_range(0..=1 << 20)?,
             linear_memory_keep_resident: u.int_in_range(0..=1 << 20)?,
 
+            decommit_batch_size: u.int_in_range(1..=1000)?,
             max_unused_warm_slots: u.int_in_range(0..=total_memories + 10)?,
 
             async_stack_zeroing: u.arbitrary()?,
diff --git a/crates/wasi-preview1-component-adapter/src/descriptors.rs b/crates/wasi-preview1-component-adapter/src/descriptors.rs
index 52b2378c7..ed4497b6f 100644
--- a/crates/wasi-preview1-component-adapter/src/descriptors.rs
+++ b/crates/wasi-preview1-component-adapter/src/descriptors.rs
@@ -1,8 +1,9 @@
 use crate::bindings::wasi::cli::{stderr, stdin, stdout};
 use crate::bindings::wasi::io::streams::{InputStream, OutputStream};
-use crate::{BlockingMode, BumpArena, ImportAlloc, TrappingUnwrap, WasmStr};
+use crate::{BlockingMode, BumpAlloc, ImportAlloc, State, TrappingUnwrap, WasmStr};
 use core::cell::{Cell, OnceCell, UnsafeCell};
 use core::mem::MaybeUninit;
+use core::num::NonZeroUsize;
 use wasi::{Errno, Fd};
 
 #[cfg(not(feature = "proxy"))]
@@ -145,21 +146,21 @@ pub struct Descriptors {
 
     /// Points to the head of a free-list of closed file descriptors.
     closed: Option<Fd>,
+}
 
-    /// Preopened directories. Initialized lazily. Access with `State::get_preopens`
-    /// to take care of initialization.
-    #[cfg(not(feature = "proxy"))]
-    preopens: Cell<Option<&'static [Preopen]>>,
+#[cfg(not(feature = "proxy"))]
+#[link(wasm_import_module = "wasi:filesystem/preopens@0.2.0")]
+extern "C" {
+    #[link_name = "get-directories"]
+    fn wasi_filesystem_get_directories(rval: *mut PreopenList);
 }
 
 impl Descriptors {
-    pub fn new(import_alloc: &ImportAlloc, arena: &BumpArena) -> Self {
+    pub fn new(state: &State) -> Self {
         let d = Descriptors {
             table: UnsafeCell::new(MaybeUninit::uninit()),
             table_len: Cell::new(0),
             closed: None,
-            #[cfg(not(feature = "proxy"))]
-            preopens: Cell::new(None),
         };
 
         fn new_once<T>(val: T) -> OnceCell<T> {
@@ -188,52 +189,71 @@ impl Descriptors {
         .trapping_unwrap();
 
         #[cfg(not(feature = "proxy"))]
-        d.open_preopens(import_alloc, arena);
+        d.open_preopens(state);
         d
     }
 
     #[cfg(not(feature = "proxy"))]
-    fn open_preopens(&self, import_alloc: &ImportAlloc, arena: &BumpArena) {
-        #[link(wasm_import_module = "wasi:filesystem/preopens@0.2.0")]
-        #[allow(improper_ctypes)] // FIXME(bytecodealliance/wit-bindgen#684)
-        extern "C" {
-            #[link_name = "get-directories"]
-            fn get_preopens_import(rval: *mut PreopenList);
+    fn open_preopens(&self, state: &State) {
+        unsafe {
+            let alloc = ImportAlloc::CountAndDiscardStrings {
+                strings_size: 0,
+                alloc: state.temporary_alloc(),
+            };
+            let (preopens, _) = state.with_import_alloc(alloc, || {
+                let mut preopens = PreopenList {
+                    base: std::ptr::null(),
+                    len: 0,
+                };
+                wasi_filesystem_get_directories(&mut preopens);
+                preopens
+            });
+            for i in 0..preopens.len {
+                let preopen = preopens.base.add(i).read();
+                // Expectation is that the descriptor index is initialized with
+                // stdio (0,1,2) and no others, so that preopens are 3..
+                let descriptor_type = preopen.descriptor.get_type().trapping_unwrap();
+                self.push(Descriptor::Streams(Streams {
+                    input: OnceCell::new(),
+                    output: OnceCell::new(),
+                    type_: StreamType::File(File {
+                        fd: preopen.descriptor,
+                        descriptor_type,
+                        position: Cell::new(0),
+                        append: false,
+                        blocking_mode: BlockingMode::Blocking,
+                        preopen_name_len: NonZeroUsize::new(preopen.path.len),
+                    }),
+                }))
+                .trapping_unwrap();
+            }
         }
-        let mut list = PreopenList {
-            base: std::ptr::null(),
-            len: 0,
+    }
+
+    #[cfg(not(feature = "proxy"))]
+    pub unsafe fn get_preopen_path(&self, state: &State, fd: Fd, path: *mut u8, len: usize) {
+        let alloc = ImportAlloc::GetPreopenPath {
+            path: Some(BumpAlloc { base: path, len }),
+            nth: fd - 3,
+            alloc: state.temporary_alloc(),
         };
-        import_alloc.with_arena(arena, || unsafe {
-            get_preopens_import(&mut list as *mut _)
+        let (preopens, _) = state.with_import_alloc(alloc, || {
+            let mut preopens = PreopenList {
+                base: std::ptr::null(),
+                len: 0,
+            };
+            wasi_filesystem_get_directories(&mut preopens);
+            preopens
         });
-        let preopens: &'static [Preopen] = unsafe {
-            // allocation comes from long lived arena, so it is safe to
-            // cast this to a &'static slice:
-            std::slice::from_raw_parts(list.base, list.len)
-        };
-        for preopen in preopens {
-            // Acquire ownership of the descriptor, leaving the rest of the
-            // `Preopen` struct in place.
-            let descriptor = unsafe { preopen.descriptor.assume_init_read() };
-            // Expectation is that the descriptor index is initialized with
-            // stdio (0,1,2) and no others, so that preopens are 3..
-            let descriptor_type = descriptor.get_type().trapping_unwrap();
-            self.push(Descriptor::Streams(Streams {
-                input: OnceCell::new(),
-                output: OnceCell::new(),
-                type_: StreamType::File(File {
-                    fd: descriptor,
-                    descriptor_type,
-                    position: Cell::new(0),
-                    append: false,
-                    blocking_mode: BlockingMode::Blocking,
-                }),
-            }))
-            .trapping_unwrap();
-        }
 
-        self.preopens.set(Some(preopens));
+        // NB: we just got owned handles for all preopened directories. We're
+        // only interested in one individual string allocation, however, so
+        // discard all of the descriptors and close them since we otherwise
+        // don't want to leak them.
+        for i in 0..preopens.len {
+            let preopen = preopens.base.add(i).read();
+            drop(preopen.descriptor);
+        }
     }
 
     fn push(&self, desc: Descriptor) -> Result<Fd, Errno> {
@@ -299,14 +319,6 @@ impl Descriptors {
             .ok_or(wasi::ERRNO_BADF)
     }
 
-    #[cfg(not(feature = "proxy"))]
-    pub fn get_preopen(&self, fd: Fd) -> Option<&Preopen> {
-        let preopens = self.preopens.get().trapping_unwrap();
-        // Subtract 3 for the stdio indices to compute the preopen index.
-        let index = fd.checked_sub(3)? as usize;
-        preopens.get(index)
-    }
-
     // Internal: close a fd, returning the descriptor.
     fn close_(&mut self, fd: Fd) -> Result<Descriptor, Errno> {
         // Throw an error if closing an fd which is already closed
@@ -437,9 +449,7 @@ impl Descriptors {
 #[cfg(not(feature = "proxy"))]
 #[repr(C)]
 pub struct Preopen {
-    /// This is `MaybeUninit` because we take ownership of the `Descriptor` to
-    /// put it in our own table.
-    pub descriptor: MaybeUninit<filesystem::Descriptor>,
+    pub descriptor: filesystem::Descriptor,
     pub path: WasmStr,
 }
 
diff --git a/crates/wasi-preview1-component-adapter/src/lib.rs b/crates/wasi-preview1-component-adapter/src/lib.rs
index 00f898bc2..8fd81c1bc 100644
--- a/crates/wasi-preview1-component-adapter/src/lib.rs
+++ b/crates/wasi-preview1-component-adapter/src/lib.rs
@@ -22,6 +22,7 @@ use core::cmp::min;
 use core::ffi::c_void;
 use core::hint::black_box;
 use core::mem::{self, align_of, forget, size_of, ManuallyDrop, MaybeUninit};
+use core::num::NonZeroUsize;
 use core::ops::{Deref, DerefMut};
 use core::ptr::{self, null_mut};
 use core::slice;
@@ -185,168 +186,229 @@ pub unsafe extern "C" fn cabi_import_realloc(
     }
     let mut ptr = null_mut::<u8>();
     State::with(|state| {
-        ptr = state.import_alloc.alloc(align, new_size);
+        let mut alloc = state.import_alloc.replace(ImportAlloc::None);
+        ptr = alloc.alloc(align, new_size);
+        state.import_alloc.set(alloc);
         Ok(())
     });
     ptr
 }
 
-/// Bump-allocated memory arena. This is a singleton - the
-/// memory will be sized according to `bump_arena_size()`.
-pub struct BumpArena {
-    data: MaybeUninit<[u8; bump_arena_size()]>,
-    position: Cell<usize>,
+/// Different ways that calling imports can allocate memory.
+///
+/// This behavior is used to customize the behavior of `cabi_import_realloc`.
+/// This is configured within `State` whenever an import is called that may
+/// invoke `cabi_import_realloc`.
+///
+/// The general idea behind these various behaviors of import allocation is
+/// that we're limited for space in the adapter here to 1 page of memory but
+/// that may not fit the total sum of arguments, environment variables, and
+/// preopens. WASIp1 APIs all provide user-provided buffers as well for these
+/// allocations so we technically don't need to store them in the adapter
+/// itself. Instead what this does is it tries to copy strings and such directly
+/// into their destination pointers where possible.
+///
+/// The types requiring allocation in the WASIp2 APIs that the WASIp1 APIs call
+/// are relatively simple. They all look like `list<T>` where `T` only has
+/// indirections in the form of `String`. This means that we can apply a
+/// "clever" hack where the alignment of an allocation is used to disambiguate
+/// whether we're allocating a string or allocating the `list<T>` allocation.
+/// This signal with alignment means that we can configure where everything
+/// goes.
+///
+/// For example consider `args_sizes_get` and `args_get`. When `args_sizes_get`
+/// is called the `list<T>` allocation happens first with alignment 4. This
+/// must be valid for the rest of the strings since the canonical ABI will fill
+/// it in, so it's allocated from `State::temporary_data`. Next all other
+/// arguments will be `string` type with alignment 1. These are also allocated
+/// within `State::temporary_data` but they're "allocated on top of one
+/// another" meaning that internal allocator state isn't updated after a string
+/// is allocated. While these strings are discarded their sizes are all summed
+/// up and returned from `args_sizes_get`.
+///
+/// Later though when `args_get` is called it's a similar allocation strategy
+/// except that strings are instead redirected to the allocation provided to
+/// `args_get` itself. This enables strings to be directly allocated into their
+/// destinations.
+///
+/// Overall this means that we're limiting the maximum number of arguments plus
+/// the size of the largest string, but otherwise we're not limiting the total
+/// size of all arguments (or env vars, preopens, etc).
+enum ImportAlloc {
+    /// A single allocation from the provided `BumpAlloc` is supported. After
+    /// the single allocation is performed all future allocations will fail.
+    OneAlloc(BumpAlloc),
+
+    /// An allocator intended for `list<T>` where `T` has string types but no
+    /// other indirections. String allocations are discarded but counted for
+    /// size.
+    ///
+    /// This allocator will use `alloc` for all allocations. Any string-related
+    /// allocation, detected via an alignment of 1, is considered "temporary"
+    /// and doesn't affect the internal state of the allocator. The allocation
+    /// is assumed to not be used after the import call returns.
+    ///
+    /// The total sum of all string sizes, however, is accumulated within
+    /// `strings_size`.
+    CountAndDiscardStrings {
+        strings_size: usize,
+        alloc: BumpAlloc,
+    },
+
+    /// An allocator intended for `list<T>` where `T` has string types but no
+    /// other indirections. String allocations go into `strings` and the
+    /// `list<..>` allocation goes into `pointers`.
+    ///
+    /// This allocator enables placing strings within a caller-supplied buffer
+    /// configured with `strings`. The `pointers` allocation is
+    /// `State::temporary_data`.
+    ///
+    /// This will additionally over-allocate strings with one extra byte to be
+    /// nul-terminated or `=`-terminated in the case of env vars.
+    SeparateStringsAndPointers {
+        strings: BumpAlloc,
+        pointers: BumpAlloc,
+    },
+
+    /// An allocator specifically for getting the nth string allocation used
+    /// for preopens.
+    ///
+    /// This will allocate the `nth` string allocation within the `path`
+    /// allocator. Otherwise all string allocations are discarded but
+    /// temoprarily placed into `alloc`.
+    GetPreopenPath {
+        path: Option<BumpAlloc>,
+        nth: u32,
+        alloc: BumpAlloc,
+    },
+
+    /// No import allocator is configured and if an allocation happens then
+    /// this will abort.
+    None,
 }
 
-impl BumpArena {
-    fn new() -> Self {
-        BumpArena {
-            data: MaybeUninit::uninit(),
-            position: Cell::new(0),
-        }
-    }
-    fn alloc(&self, align: usize, size: usize) -> *mut u8 {
-        let start = self.data.as_ptr() as usize;
-        let next = start + self.position.get();
-        let alloc = align_to(next, align);
-        let offset = alloc - start;
-        if offset + size > bump_arena_size() {
-            unreachable!("out of memory");
+impl ImportAlloc {
+    /// To be used by cabi_import_realloc only!
+    unsafe fn alloc(&mut self, align: usize, size: usize) -> *mut u8 {
+        match self {
+            ImportAlloc::OneAlloc(alloc) => {
+                let ret = alloc.alloc(align, size);
+                *self = ImportAlloc::None;
+                ret
+            }
+            ImportAlloc::SeparateStringsAndPointers { strings, pointers } => {
+                if align == 1 {
+                    strings.alloc(align, size + 1)
+                } else {
+                    pointers.alloc(align, size)
+                }
+            }
+            ImportAlloc::CountAndDiscardStrings {
+                strings_size,
+                alloc,
+            } => {
+                if align == 1 {
+                    *strings_size += size;
+                    alloc.clone().alloc(align, size)
+                } else {
+                    alloc.alloc(align, size)
+                }
+            }
+            ImportAlloc::GetPreopenPath { path, nth, alloc } => {
+                if align == 1 {
+                    if *nth == 0 {
+                        if let Some(a) = path {
+                            let ret = a.alloc(align, size);
+                            *path = None;
+                            return ret;
+                        }
+                    } else {
+                        *nth -= 1;
+                    }
+                    alloc.clone().alloc(align, size)
+                } else {
+                    alloc.alloc(align, size)
+                }
+            }
+            ImportAlloc::None => unreachable!("no allocator configured"),
         }
-        self.position.set(offset + size);
-        alloc as *mut u8
     }
 }
-fn align_to(ptr: usize, align: usize) -> usize {
-    (ptr + (align - 1)) & !(align - 1)
-}
 
-// Invariant: buffer not-null and arena is-some are never true at the same
-// time. We did not use an enum to make this invalid behavior unrepresentable
-// because we can't use RefCell to borrow() the variants of the enum - only
-// Cell provides mutability without pulling in panic machinery - so it would
-// make the accessors a lot more awkward to write.
-pub struct ImportAlloc {
-    // When not-null, allocator should use this buffer/len pair at most once
-    // to satisfy allocations.
-    buffer: Cell<*mut u8>,
-    len: Cell<usize>,
-    // When not-empty, allocator should use this arena to satisfy allocations.
-    arena: Cell<Option<&'static BumpArena>>,
+/// Helper type to manage allocations from a `base`/`len` combo.
+///
+/// This isn't really used much in an arena-style per se but it's used in
+/// combination with the `ImportAlloc` flavors above.
+#[derive(Clone)]
+struct BumpAlloc {
+    base: *mut u8,
+    len: usize,
 }
 
-impl ImportAlloc {
-    fn new() -> Self {
-        ImportAlloc {
-            buffer: Cell::new(std::ptr::null_mut()),
-            len: Cell::new(0),
-            arena: Cell::new(None),
+impl BumpAlloc {
+    unsafe fn alloc(&mut self, align: usize, size: usize) -> *mut u8 {
+        self.align_to(align);
+        if size > self.len {
+            unreachable!("allocation size is too large")
         }
+        self.len -= size;
+        let ret = self.base;
+        self.base = ret.add(size);
+        ret
     }
 
-    /// Expect at most one import allocation during execution of the provided closure.
-    /// Use the provided buffer to satisfy that import allocation. The user is responsible
-    /// for making sure allocated imports are not used beyond the lifetime of the buffer.
-    fn with_buffer<T>(&self, buffer: *mut u8, len: usize, f: impl FnOnce() -> T) -> T {
-        if self.arena.get().is_some() {
-            unreachable!("arena mode")
-        }
-        let prev = self.buffer.replace(buffer);
-        if !prev.is_null() {
-            unreachable!("overwrote another buffer")
-        }
-        self.len.set(len);
-        let r = f();
-        self.buffer.set(std::ptr::null_mut());
-        r
-    }
-
-    /// Permit many import allocations during execution of the provided closure.
-    /// Use the provided BumpArena to satisfry those allocations. The user is responsible
-    /// for making sure allocated imports are not used beyond the lifetime of the arena.
-    fn with_arena<T>(&self, arena: &BumpArena, f: impl FnOnce() -> T) -> T {
-        if !self.buffer.get().is_null() {
-            unreachable!("buffer mode")
+    unsafe fn align_to(&mut self, align: usize) {
+        if !align.is_power_of_two() {
+            unreachable!("invalid alignment");
         }
-        let prev = self.arena.replace(Some(unsafe {
-            // Safety: Need to erase the lifetime to store in the arena cell.
-            std::mem::transmute::<&'_ BumpArena, &'static BumpArena>(arena)
-        }));
-        if prev.is_some() {
-            unreachable!("overwrote another arena")
-        }
-        let r = f();
-        self.arena.set(None);
-        r
-    }
-
-    /// To be used by cabi_import_realloc only!
-    fn alloc(&self, align: usize, size: usize) -> *mut u8 {
-        if let Some(arena) = self.arena.get() {
-            arena.alloc(align, size)
-        } else {
-            let buffer = self.buffer.get();
-            if buffer.is_null() {
-                unreachable!("buffer not provided, or already used")
-            }
-            let buffer = buffer as usize;
-            let alloc = align_to(buffer, align);
-            if alloc.checked_add(size).trapping_unwrap()
-                > buffer.checked_add(self.len.get()).trapping_unwrap()
-            {
-                unreachable!("out of memory")
-            }
-            self.buffer.set(std::ptr::null_mut());
-            alloc as *mut u8
+        let align_offset = self.base.align_offset(align);
+        if align_offset >= self.len {
+            unreachable!("failed to allocate")
         }
+        self.len -= align_offset;
+        self.base = self.base.add(align_offset);
     }
 }
 
-/// This allocator is only used for the `run` entrypoint.
-///
-/// The implementation here is a bump allocator into `State::long_lived_arena` which
-/// traps when it runs out of data. This means that the total size of
-/// arguments/env/etc coming into a component is bounded by the current 64k
-/// (ish) limit. That's just an implementation limit though which can be lifted
-/// by dynamically calling the main module's allocator as necessary for more data.
-#[no_mangle]
-pub unsafe extern "C" fn cabi_export_realloc(
-    old_ptr: *mut u8,
-    old_size: usize,
-    align: usize,
-    new_size: usize,
-) -> *mut u8 {
-    if !old_ptr.is_null() || old_size != 0 {
-        unreachable!();
-    }
-    let mut ret = null_mut::<u8>();
-    State::with(|state| {
-        ret = state.long_lived_arena.alloc(align, new_size);
-        Ok(())
-    });
-    ret
+#[cfg(not(feature = "proxy"))]
+#[link(wasm_import_module = "wasi:cli/environment@0.2.0")]
+extern "C" {
+    #[link_name = "get-arguments"]
+    fn wasi_cli_get_arguments(rval: *mut WasmStrList);
+    #[link_name = "get-environment"]
+    fn wasi_cli_get_environment(rval: *mut StrTupleList);
 }
 
 /// Read command-line argument data.
 /// The size of the array should match that returned by `args_sizes_get`
 #[no_mangle]
-pub unsafe extern "C" fn args_get(mut argv: *mut *mut u8, mut argv_buf: *mut u8) -> Errno {
+pub unsafe extern "C" fn args_get(argv: *mut *mut u8, argv_buf: *mut u8) -> Errno {
     State::with(|state| {
         #[cfg(not(feature = "proxy"))]
         {
-            for arg in state.get_args() {
-                // Copy the argument into `argv_buf` which must be sized
-                // appropriately by the caller.
-                ptr::copy_nonoverlapping(arg.ptr, argv_buf, arg.len);
-                *argv_buf.add(arg.len) = 0;
-
-                // Copy the argument pointer into the `argv` buf
-                *argv = argv_buf;
-
-                // Update our pointers past what's written to prepare for the
-                // next argument.
-                argv = argv.add(1);
-                argv_buf = argv_buf.add(arg.len + 1);
+            let alloc = ImportAlloc::SeparateStringsAndPointers {
+                strings: BumpAlloc {
+                    base: argv_buf,
+                    len: usize::MAX,
+                },
+                pointers: state.temporary_alloc(),
+            };
+            let (list, _) = state.with_import_alloc(alloc, || unsafe {
+                let mut list = WasmStrList {
+                    base: std::ptr::null(),
+                    len: 0,
+                };
+                wasi_cli_get_arguments(&mut list);
+                list
+            });
+
+            // Fill in `argv` by walking over the returned `list` and then
+            // additionally apply the nul-termination for each argument itself
+            // here.
+            for i in 0..list.len {
+                let s = list.base.add(i).read();
+                *argv.add(i) = s.ptr.cast_mut();
+                *s.ptr.add(s.len).cast_mut() = 0;
             }
         }
         Ok(())
@@ -364,11 +426,30 @@ pub unsafe extern "C" fn args_sizes_get(argc: *mut Size, argv_buf_size: *mut Siz
         }
         #[cfg(not(feature = "proxy"))]
         {
-            let args = state.get_args();
-            *argc = args.len();
-            // Add one to each length for the terminating nul byte added by
-            // the `args_get` function.
-            *argv_buf_size = args.iter().map(|s| s.len + 1).sum();
+            let alloc = ImportAlloc::CountAndDiscardStrings {
+                strings_size: 0,
+                alloc: state.temporary_alloc(),
+            };
+            let (len, alloc) = state.with_import_alloc(alloc, || unsafe {
+                let mut list = WasmStrList {
+                    base: std::ptr::null(),
+                    len: 0,
+                };
+                wasi_cli_get_arguments(&mut list);
+                list.len
+            });
+            match alloc {
+                ImportAlloc::CountAndDiscardStrings {
+                    strings_size,
+                    alloc: _,
+                } => {
+                    *argc = len;
+                    // add in bytes needed for a 0-byte at the end of each
+                    // argument.
+                    *argv_buf_size = strings_size + len;
+                }
+                _ => unreachable!(),
+            }
         }
         Ok(())
     })
@@ -377,27 +458,35 @@ pub unsafe extern "C" fn args_sizes_get(argc: *mut Size, argv_buf_size: *mut Siz
 /// Read environment variable data.
 /// The sizes of the buffers should match that returned by `environ_sizes_get`.
 #[no_mangle]
-pub unsafe extern "C" fn environ_get(environ: *mut *mut u8, environ_buf: *mut u8) -> Errno {
+pub unsafe extern "C" fn environ_get(environ: *mut *const u8, environ_buf: *mut u8) -> Errno {
     State::with(|state| {
         #[cfg(not(feature = "proxy"))]
         {
-            let mut offsets = environ;
-            let mut buffer = environ_buf;
-            for var in state.get_environment() {
-                ptr::write(offsets, buffer);
-                offsets = offsets.add(1);
-
-                ptr::copy_nonoverlapping(var.key.ptr, buffer, var.key.len);
-                buffer = buffer.add(var.key.len);
-
-                ptr::write(buffer, b'=');
-                buffer = buffer.add(1);
-
-                ptr::copy_nonoverlapping(var.value.ptr, buffer, var.value.len);
-                buffer = buffer.add(var.value.len);
+            let alloc = ImportAlloc::SeparateStringsAndPointers {
+                strings: BumpAlloc {
+                    base: environ_buf,
+                    len: usize::MAX,
+                },
+                pointers: state.temporary_alloc(),
+            };
+            let (list, _) = state.with_import_alloc(alloc, || unsafe {
+                let mut list = StrTupleList {
+                    base: std::ptr::null(),
+                    len: 0,
+                };
+                wasi_cli_get_environment(&mut list);
+                list
+            });
 
-                ptr::write(buffer, 0);
-                buffer = buffer.add(1);
+            // Fill in `environ` by walking over the returned `list`. Strings
+            // are guaranteed to be allocated next to each other with one
+            // extra byte at the end, so also insert the `=` between keys and
+            // the `\0` at the end of the env var.
+            for i in 0..list.len {
+                let s = list.base.add(i).read();
+                *environ.add(i) = s.key.ptr;
+                *s.key.ptr.add(s.key.len).cast_mut() = b'=';
+                *s.value.ptr.add(s.value.len).cast_mut() = 0;
             }
         }
 
@@ -428,15 +517,30 @@ pub unsafe extern "C" fn environ_sizes_get(
         }
         #[cfg(not(feature = "proxy"))]
         {
-            let vars = state.get_environment();
-            *environc = vars.len();
-            *environ_buf_size = {
-                let mut sum = 0;
-                for var in vars {
-                    sum += var.key.len + var.value.len + 2;
-                }
-                sum
+            let alloc = ImportAlloc::CountAndDiscardStrings {
+                strings_size: 0,
+                alloc: state.temporary_alloc(),
             };
+            let (len, alloc) = state.with_import_alloc(alloc, || unsafe {
+                let mut list = StrTupleList {
+                    base: std::ptr::null(),
+                    len: 0,
+                };
+                wasi_cli_get_environment(&mut list);
+                list.len
+            });
+            match alloc {
+                ImportAlloc::CountAndDiscardStrings {
+                    strings_size,
+                    alloc: _,
+                } => {
+                    *environc = len;
+                    // Account for `=` between keys and a 0-byte at the end of
+                    // each key.
+                    *environ_buf_size = strings_size + 2 * len;
+                }
+                _ => unreachable!(),
+            }
         }
 
         Ok(())
@@ -881,8 +985,7 @@ pub unsafe extern "C" fn fd_pread(
             let ds = state.descriptors();
             let file = ds.get_file(fd)?;
             let (data, end) = state
-                .import_alloc
-                .with_buffer(ptr, len, || file.fd.read(len as u64, offset))?;
+                .with_one_import_alloc(ptr, len, || file.fd.read(len as u64, offset))?;
             assert_eq!(data.as_ptr(), ptr);
             assert!(data.len() <= len);
 
@@ -918,19 +1021,26 @@ pub unsafe extern "C" fn fd_prestat_get(fd: Fd, buf: *mut Prestat) -> Errno {
     cfg_filesystem_available! {
         State::with(|state| {
             let ds = state.descriptors();
-            if let Some(preopen) = ds.get_preopen(fd) {
-                buf.write(Prestat {
-                    tag: 0,
-                    u: PrestatU {
-                        dir: PrestatDir {
-                            pr_name_len: preopen.path.len,
+            match ds.get(fd)? {
+                Descriptor::Streams(Streams {
+                    type_: StreamType::File(File {
+                        preopen_name_len: Some(len),
+                        ..
+                    }),
+                    ..
+                }) => {
+                    buf.write(Prestat {
+                        tag: 0,
+                        u: PrestatU {
+                            dir: PrestatDir {
+                                pr_name_len: len.get(),
+                            },
                         },
-                    },
-                });
+                    });
 
-                Ok(())
-            } else {
-                Err(ERRNO_BADF)
+                    Ok(())
+                }
+                _ => Err(ERRNO_BADF),
             }
         })
     }
@@ -942,16 +1052,22 @@ pub unsafe extern "C" fn fd_prestat_dir_name(fd: Fd, path: *mut u8, path_max_len
     cfg_filesystem_available! {
         State::with(|state| {
             let ds = state.descriptors();
-            if let Some(preopen) = ds.get_preopen(fd) {
-                if preopen.path.len > path_max_len {
-                    Err(ERRNO_NAMETOOLONG)
-                } else {
-                    ptr::copy_nonoverlapping(preopen.path.ptr, path, preopen.path.len);
-                    Ok(())
-                }
-            } else {
-                Err(ERRNO_NOTDIR)
+            let preopen_len = match ds.get(fd)? {
+                Descriptor::Streams(Streams {
+                    type_: StreamType::File(File {
+                        preopen_name_len: Some(len),
+                        ..
+                    }),
+                    ..
+                }) => len.get(),
+                _ => return Err(ERRNO_BADF),
+            };
+            if preopen_len > path_max_len {
+                return Err(ERRNO_NAMETOOLONG)
             }
+
+            ds.get_preopen_path(state, fd, path, path_max_len);
+            Ok(())
         })
     }
 }
@@ -1028,8 +1144,7 @@ pub unsafe extern "C" fn fd_read(
                 let read_len = u64::try_from(len).trapping_unwrap();
                 let wasi_stream = streams.get_read_stream()?;
                 let data = match state
-                    .import_alloc
-                    .with_buffer(ptr, len, || blocking_mode.read(wasi_stream, read_len))
+                    .with_one_import_alloc(ptr, len, || blocking_mode.read(wasi_stream, read_len))
                 {
                     Ok(data) => data,
                     Err(streams::StreamError::Closed) => {
@@ -1288,11 +1403,9 @@ pub unsafe extern "C" fn fd_readdir(
                     Ok((dirent, buffer))
                 });
             }
-            let entry = self.state.import_alloc.with_buffer(
-                self.state.path_buf.get().cast(),
-                PATH_MAX,
-                || self.stream.0.read_directory_entry(),
-            );
+            let entry = self
+                .state
+                .with_one_temporary_alloc(|| self.stream.0.read_directory_entry());
             let entry = match entry {
                 Ok(Some(entry)) => entry,
                 Ok(None) => return None,
@@ -1315,7 +1428,7 @@ pub unsafe extern "C" fn fd_readdir(
             // Extend the lifetime of `name` to the `self.state` lifetime for
             // this iterator since the data for the name lives within state.
             let name = unsafe {
-                assert_eq!(name.as_ptr(), self.state.path_buf.get().cast());
+                assert_eq!(name.as_ptr(), self.state.temporary_data.get().cast());
                 slice::from_raw_parts(name.as_ptr().cast(), name.len())
             };
             Some(Ok((dirent, name)))
@@ -1652,6 +1765,7 @@ pub unsafe extern "C" fn path_open(
                     } else {
                         BlockingMode::NonBlocking
                     },
+                    preopen_name_len: None,
                 }),
             });
 
@@ -1686,14 +1800,12 @@ pub unsafe extern "C" fn path_readlink(
             let file = ds.get_dir(fd)?;
             let path = if use_state_buf {
                 state
-                    .import_alloc
-                    .with_buffer(state.path_buf.get().cast(), PATH_MAX, || {
+                    .with_one_temporary_alloc( || {
                         file.fd.readlink_at(path)
                     })?
             } else {
                 state
-                    .import_alloc
-                    .with_buffer(buf, buf_len, || file.fd.readlink_at(path))?
+                    .with_one_import_alloc(buf, buf_len, || file.fd.readlink_at(path))?
             };
 
             if use_state_buf {
@@ -1960,7 +2072,7 @@ pub unsafe extern "C" fn poll_oneoff(
             len: 0,
         };
 
-        state.import_alloc.with_buffer(
+        state.with_one_import_alloc(
             results.cast(),
             nsubscriptions
                 .checked_mul(size_of::<u32>())
@@ -2106,8 +2218,7 @@ pub unsafe extern "C" fn random_get(buf: *mut u8, buf_len: Size) -> Errno {
         State::with(|state| {
             assert_eq!(buf_len as u32 as Size, buf_len);
             let result = state
-                .import_alloc
-                .with_buffer(buf, buf_len, || random::get_random_bytes(buf_len as u64));
+                .with_one_import_alloc(buf, buf_len, || random::get_random_bytes(buf_len as u64));
             assert_eq!(result.as_ptr(), buf);
 
             // The returned buffer's memory was allocated in `buf`, so don't separately
@@ -2386,6 +2497,9 @@ pub struct File {
     /// blocking_check_write on the underlying streams. When false, read and write
     /// dispatch to stream's plain read and check_write.
     blocking_mode: BlockingMode,
+
+    /// TODO
+    preopen_name_len: Option<NonZeroUsize>,
 }
 
 #[cfg(not(feature = "proxy"))]
@@ -2418,7 +2532,7 @@ struct State {
     magic1: u32,
 
     /// Used to coordinate allocations of `cabi_import_realloc`
-    import_alloc: ImportAlloc,
+    import_alloc: Cell<ImportAlloc>,
 
     /// Storage of mapping from preview1 file descriptors to preview2 file
     /// descriptors.
@@ -2427,27 +2541,8 @@ struct State {
     /// lazy initialization happens.
     descriptors: RefCell<Option<Descriptors>>,
 
-    /// Auxiliary storage to handle the `path_readlink` function.
-    #[cfg(not(feature = "proxy"))]
-    path_buf: UnsafeCell<MaybeUninit<[u8; PATH_MAX]>>,
-
-    /// Long-lived bump allocated memory arena.
-    ///
-    /// This is used for the cabi_export_realloc to allocate data passed to the
-    /// `run` entrypoint. Allocations in this arena are safe to use for
-    /// the lifetime of the State struct. It may also be used for import allocations
-    /// which need to be long-lived, by using `import_alloc.with_arena`.
-    long_lived_arena: BumpArena,
-
-    /// Arguments. Initialized lazily. Access with `State::get_args` to take care of
-    /// initialization.
-    #[cfg(not(feature = "proxy"))]
-    args: Cell<Option<&'static [WasmStr]>>,
-
-    /// Environment variables. Initialized lazily. Access with `State::get_environment`
-    /// to take care of initialization.
-    #[cfg(not(feature = "proxy"))]
-    env_vars: Cell<Option<&'static [StrTuple]>>,
+    /// TODO
+    temporary_data: UnsafeCell<MaybeUninit<[u8; temporary_data_size()]>>,
 
     /// Cache for the `fd_readdir` call for a final `wasi::Dirent` plus path
     /// name that didn't fit into the caller's buffer.
@@ -2507,7 +2602,7 @@ pub struct ReadyList {
     len: usize,
 }
 
-const fn bump_arena_size() -> usize {
+const fn temporary_data_size() -> usize {
     // The total size of the struct should be a page, so start there
     let mut start = PAGE_SIZE;
 
@@ -2515,12 +2610,11 @@ const fn bump_arena_size() -> usize {
     start -= size_of::<Descriptors>();
     #[cfg(not(feature = "proxy"))]
     {
-        start -= PATH_MAX;
         start -= size_of::<DirentCache>();
     }
 
     // Remove miscellaneous metadata also stored in state.
-    let misc = if cfg!(feature = "proxy") { 7 } else { 14 };
+    let misc = if cfg!(feature = "proxy") { 9 } else { 12 };
     start -= misc * size_of::<usize>();
 
     // Everything else is the `command_data` allocation.
@@ -2618,15 +2712,9 @@ impl State {
         state.write(State {
             magic1: MAGIC,
             magic2: MAGIC,
-            import_alloc: ImportAlloc::new(),
+            import_alloc: Cell::new(ImportAlloc::None),
             descriptors: RefCell::new(None),
-            #[cfg(not(feature = "proxy"))]
-            path_buf: UnsafeCell::new(MaybeUninit::uninit()),
-            long_lived_arena: BumpArena::new(),
-            #[cfg(not(feature = "proxy"))]
-            args: Cell::new(None),
-            #[cfg(not(feature = "proxy"))]
-            env_vars: Cell::new(None),
+            temporary_data: UnsafeCell::new(MaybeUninit::uninit()),
             #[cfg(not(feature = "proxy"))]
             dirent_cache: DirentCache {
                 stream: Cell::new(None),
@@ -2652,7 +2740,7 @@ impl State {
             .try_borrow_mut()
             .unwrap_or_else(|_| unreachable!());
         if d.is_none() {
-            *d = Some(Descriptors::new(&self.import_alloc, &self.long_lived_arena));
+            *d = Some(Descriptors::new(self));
         }
         RefMut::map(d, |d| d.as_mut().unwrap_or_else(|| unreachable!()))
     }
@@ -2664,58 +2752,47 @@ impl State {
             .try_borrow_mut()
             .unwrap_or_else(|_| unreachable!());
         if d.is_none() {
-            *d = Some(Descriptors::new(&self.import_alloc, &self.long_lived_arena));
+            *d = Some(Descriptors::new(self));
         }
         RefMut::map(d, |d| d.as_mut().unwrap_or_else(|| unreachable!()))
     }
 
-    #[cfg(not(feature = "proxy"))]
-    fn get_environment(&self) -> &[StrTuple] {
-        if self.env_vars.get().is_none() {
-            #[link(wasm_import_module = "wasi:cli/environment@0.2.0")]
-            extern "C" {
-                #[link_name = "get-environment"]
-                fn get_environment_import(rval: *mut StrTupleList);
-            }
-            let mut list = StrTupleList {
-                base: std::ptr::null(),
-                len: 0,
-            };
-            self.import_alloc
-                .with_arena(&self.long_lived_arena, || unsafe {
-                    get_environment_import(&mut list as *mut _)
-                });
-            self.env_vars.set(Some(unsafe {
-                /* allocation comes from long lived arena, so it is safe to
-                 * cast this to a &'static slice: */
-                std::slice::from_raw_parts(list.base, list.len)
-            }));
+    unsafe fn temporary_alloc(&self) -> BumpAlloc {
+        BumpAlloc {
+            base: self.temporary_data.get().cast(),
+            len: mem::size_of_val(&self.temporary_data),
         }
-        self.env_vars.get().trapping_unwrap()
     }
 
-    #[cfg(not(feature = "proxy"))]
-    fn get_args(&self) -> &[WasmStr] {
-        if self.args.get().is_none() {
-            #[link(wasm_import_module = "wasi:cli/environment@0.2.0")]
-            extern "C" {
-                #[link_name = "get-arguments"]
-                fn get_args_import(rval: *mut WasmStrList);
-            }
-            let mut list = WasmStrList {
-                base: std::ptr::null(),
-                len: 0,
-            };
-            self.import_alloc
-                .with_arena(&self.long_lived_arena, || unsafe {
-                    get_args_import(&mut list as *mut _)
-                });
-            self.args.set(Some(unsafe {
-                /* allocation comes from long lived arena, so it is safe to
-                 * cast this to a &'static slice: */
-                std::slice::from_raw_parts(list.base, list.len)
-            }));
+    /// Configure that `cabi_import_realloc` will allocate once from
+    /// `self.temporary_data` for the duration of the closure `f`.
+    ///
+    /// Panics if the import allocator is already configured.
+    fn with_one_temporary_alloc<T>(&self, f: impl FnOnce() -> T) -> T {
+        let alloc = unsafe { self.temporary_alloc() };
+        self.with_import_alloc(ImportAlloc::OneAlloc(alloc), f).0
+    }
+
+    /// Configure that `cabi_import_realloc` will allocate once from
+    /// `base` with at most `len` bytes for the duration of `f`.
+    ///
+    /// Panics if the import allocator is already configured.
+    fn with_one_import_alloc<T>(&self, base: *mut u8, len: usize, f: impl FnOnce() -> T) -> T {
+        let alloc = BumpAlloc { base, len };
+        self.with_import_alloc(ImportAlloc::OneAlloc(alloc), f).0
+    }
+
+    /// Configures the `alloc` specified to be the allocator for
+    /// `cabi_import_realloc` for the duration of `f`.
+    ///
+    /// Panics if the import allocator is already configured.
+    fn with_import_alloc<T>(&self, alloc: ImportAlloc, f: impl FnOnce() -> T) -> (T, ImportAlloc) {
+        match self.import_alloc.replace(alloc) {
+            ImportAlloc::None => {}
+            _ => unreachable!("import allocator already set"),
         }
-        self.args.get().trapping_unwrap()
+        let r = f();
+        let alloc = self.import_alloc.replace(ImportAlloc::None);
+        (r, alloc)
     }
 }
diff --git a/crates/wasi/src/filesystem.rs b/crates/wasi/src/filesystem.rs
index 197b2fe88..6b516ef02 100644
--- a/crates/wasi/src/filesystem.rs
+++ b/crates/wasi/src/filesystem.rs
@@ -248,6 +248,7 @@ impl FileInputStream {
     pub async fn read(&mut self, size: usize) -> Result<Bytes, StreamError> {
         use system_interface::fs::FileIoExt;
         let p = self.position;
+
         let (r, mut buf) = self
             .file
             .spawn_blocking(move |f| {
@@ -256,7 +257,7 @@ impl FileInputStream {
                 (r, buf)
             })
             .await;
-        let n = read_result(r)?;
+        let n = read_result(r, size)?;
         buf.truncate(n);
         self.position += n as u64;
         Ok(buf.freeze())
@@ -268,9 +269,9 @@ impl FileInputStream {
     }
 }
 
-fn read_result(r: io::Result<usize>) -> Result<usize, StreamError> {
+fn read_result(r: io::Result<usize>, size: usize) -> Result<usize, StreamError> {
     match r {
-        Ok(0) => Err(StreamError::Closed),
+        Ok(0) if size > 0 => Err(StreamError::Closed),
         Ok(n) => Ok(n),
         Err(e) if e.kind() == std::io::ErrorKind::Interrupted => Ok(0),
         Err(e) => Err(StreamError::LastOperationFailed(e.into())),
diff --git a/crates/wasmtime/src/compile.rs b/crates/wasmtime/src/compile.rs
index 65449702d..62f910d86 100644
--- a/crates/wasmtime/src/compile.rs
+++ b/crates/wasmtime/src/compile.rs
@@ -807,7 +807,9 @@ impl FunctionIndices {
                 // Attempt to convert table initializer segments to
                 // FuncTable representation where possible, to enable
                 // table lazy init.
-                translation.try_func_table_init();
+                if engine.tunables().table_lazy_init {
+                    translation.try_func_table_init();
+                }
 
                 let funcs: PrimaryMap<DefinedFuncIndex, CompiledFunctionInfo> =
                     wasm_functions_for_module(&mut wasm_functions, module)
diff --git a/crates/wasmtime/src/config.rs b/crates/wasmtime/src/config.rs
index 7c7da779c..b9d1448e8 100644
--- a/crates/wasmtime/src/config.rs
+++ b/crates/wasmtime/src/config.rs
@@ -149,6 +149,7 @@ struct ConfigTunables {
     epoch_interruption: Option<bool>,
     static_memory_bound_is_maximum: Option<bool>,
     guard_before_linear_memory: Option<bool>,
+    table_lazy_init: Option<bool>,
     generate_address_map: Option<bool>,
     debug_adapter_modules: Option<bool>,
     relaxed_simd_deterministic: Option<bool>,
@@ -1541,6 +1542,19 @@ impl Config {
         self
     }
 
+    /// Indicates whether to initialize tables lazily, so that instantiation
+    /// is fast but indirect calls are a little slower. If false, tables
+    /// are initialized eagerly during instantiation from any active element
+    /// segments that apply to them.
+    ///
+    /// ## Default
+    ///
+    /// This value defaults to `true`.
+    pub fn table_lazy_init(&mut self, table_lazy_init: bool) -> &mut Self {
+        self.tunables.table_lazy_init = Some(table_lazy_init);
+        self
+    }
+
     /// Configure the version information used in serialized and deserialzied [`crate::Module`]s.
     /// This effects the behavior of [`crate::Module::serialize()`], as well as
     /// [`crate::Module::deserialize()`] and related functions.
@@ -1810,6 +1824,7 @@ impl Config {
             epoch_interruption
             static_memory_bound_is_maximum
             guard_before_linear_memory
+            table_lazy_init
             generate_address_map
             debug_adapter_modules
             relaxed_simd_deterministic
@@ -1830,6 +1845,10 @@ impl Config {
             if tunables.winch_callable && tunables.tail_callable {
                 bail!("Winch does not support the WebAssembly tail call proposal");
             }
+
+            if tunables.winch_callable && !tunables.table_lazy_init {
+                bail!("Winch requires the table-lazy-init configuration option");
+            }
         }
 
         if tunables.static_memory_offset_guard_size < tunables.dynamic_memory_offset_guard_size {
@@ -2405,6 +2424,20 @@ impl PoolingAllocationConfig {
         self
     }
 
+    /// The target number of decommits to do per batch.
+    ///
+    /// This is not precise, as we can queue up decommits at times when we
+    /// aren't prepared to immediately flush them, and so we may go over this
+    /// target size occasionally.
+    ///
+    /// A batch size of one effectively disables batching.
+    ///
+    /// Defaults to `1`.
+    pub fn decommit_batch_size(&mut self, batch_size: usize) -> &mut Self {
+        self.config.decommit_batch_size = batch_size;
+        self
+    }
+
     /// Configures whether or not stacks used for async futures are reset to
     /// zero after usage.
     ///
diff --git a/crates/wasmtime/src/engine/serialization.rs b/crates/wasmtime/src/engine/serialization.rs
index 1415aa19a..e62ceb606 100644
--- a/crates/wasmtime/src/engine/serialization.rs
+++ b/crates/wasmtime/src/engine/serialization.rs
@@ -355,6 +355,7 @@ impl Metadata<'_> {
             epoch_interruption,
             static_memory_bound_is_maximum,
             guard_before_linear_memory,
+            table_lazy_init,
             relaxed_simd_deterministic,
             tail_callable,
             winch_callable,
@@ -415,6 +416,7 @@ impl Metadata<'_> {
             other.guard_before_linear_memory,
             "guard before linear memory",
         )?;
+        Self::check_bool(table_lazy_init, other.table_lazy_init, "table lazy init")?;
         Self::check_bool(
             relaxed_simd_deterministic,
             other.relaxed_simd_deterministic,
diff --git a/crates/wasmtime/src/runtime.rs b/crates/wasmtime/src/runtime.rs
index b3a5ecd52..0750ad424 100644
--- a/crates/wasmtime/src/runtime.rs
+++ b/crates/wasmtime/src/runtime.rs
@@ -61,6 +61,9 @@ pub use values::*;
 
 pub(crate) use uninhabited::*;
 
+#[cfg(feature = "pooling-allocator")]
+pub use vm::PoolConcurrencyLimitError;
+
 #[cfg(feature = "profiling")]
 mod profiling;
 #[cfg(feature = "profiling")]
diff --git a/crates/wasmtime/src/runtime/store.rs b/crates/wasmtime/src/runtime/store.rs
index 599960ebd..e84491ab3 100644
--- a/crates/wasmtime/src/runtime/store.rs
+++ b/crates/wasmtime/src/runtime/store.rs
@@ -2067,7 +2067,7 @@ impl<T> StoreContextMut<'_, T> {
             // wrap that in a custom future implementation which does the
             // translation from the future protocol to our fiber API.
             FiberFuture {
-                fiber,
+                fiber: Some(fiber),
                 current_poll_cx,
                 engine,
                 state: Some(crate::runtime::vm::AsyncWasmCallState::new()),
@@ -2078,7 +2078,7 @@ impl<T> StoreContextMut<'_, T> {
         return Ok(slot.unwrap());
 
         struct FiberFuture<'a> {
-            fiber: wasmtime_fiber::Fiber<'a, Result<()>, (), Result<()>>,
+            fiber: Option<wasmtime_fiber::Fiber<'a, Result<()>, (), Result<()>>>,
             current_poll_cx: *mut *mut Context<'static>,
             engine: Engine,
             // See comments in `FiberFuture::resume` for this
@@ -2149,6 +2149,10 @@ impl<T> StoreContextMut<'_, T> {
         unsafe impl Send for FiberFuture<'_> {}
 
         impl FiberFuture<'_> {
+            fn fiber(&self) -> &wasmtime_fiber::Fiber<'_, Result<()>, (), Result<()>> {
+                self.fiber.as_ref().unwrap()
+            }
+
             /// This is a helper function to call `resume` on the underlying
             /// fiber while correctly managing Wasmtime's thread-local data.
             ///
@@ -2174,7 +2178,7 @@ impl<T> StoreContextMut<'_, T> {
                         fiber: self,
                         state: Some(prev),
                     };
-                    return restore.fiber.fiber.resume(val);
+                    return restore.fiber.fiber().resume(val);
                 }
 
                 struct Restore<'a, 'b> {
@@ -2241,7 +2245,7 @@ impl<T> StoreContextMut<'_, T> {
                         // then that's a bug indicating that TLS management in
                         // Wasmtime is incorrect.
                         Err(()) => {
-                            if let Some(range) = self.fiber.stack().range() {
+                            if let Some(range) = self.fiber().stack().range() {
                                 crate::runtime::vm::AsyncWasmCallState::assert_current_state_not_in_range(range);
                             }
                             Poll::Pending
@@ -2268,7 +2272,7 @@ impl<T> StoreContextMut<'_, T> {
         // completion.
         impl Drop for FiberFuture<'_> {
             fn drop(&mut self) {
-                if !self.fiber.done() {
+                if !self.fiber().done() {
                     let result = self.resume(Err(anyhow!("future dropped")));
                     // This resumption with an error should always complete the
                     // fiber. While it's technically possible for host code to catch
@@ -2282,7 +2286,7 @@ impl<T> StoreContextMut<'_, T> {
                 unsafe {
                     self.engine
                         .allocator()
-                        .deallocate_fiber_stack(self.fiber.stack());
+                        .deallocate_fiber_stack(self.fiber.take().unwrap().into_stack());
                 }
             }
         }
diff --git a/crates/wasmtime/src/runtime/trampoline/memory.rs b/crates/wasmtime/src/runtime/trampoline/memory.rs
index 981f769dd..d2d85bdf1 100644
--- a/crates/wasmtime/src/runtime/trampoline/memory.rs
+++ b/crates/wasmtime/src/runtime/trampoline/memory.rs
@@ -248,7 +248,7 @@ unsafe impl InstanceAllocatorImpl for SingleMemoryInstance<'_> {
     }
 
     #[cfg(feature = "async")]
-    unsafe fn deallocate_fiber_stack(&self, _stack: &wasmtime_fiber::FiberStack) {
+    unsafe fn deallocate_fiber_stack(&self, _stack: wasmtime_fiber::FiberStack) {
         unreachable!()
     }
 
diff --git a/crates/wasmtime/src/runtime/vm.rs b/crates/wasmtime/src/runtime/vm.rs
index 95e85777b..3683b3182 100644
--- a/crates/wasmtime/src/runtime/vm.rs
+++ b/crates/wasmtime/src/runtime/vm.rs
@@ -55,7 +55,8 @@ pub use crate::runtime::vm::instance::{
 };
 #[cfg(feature = "pooling-allocator")]
 pub use crate::runtime::vm::instance::{
-    InstanceLimits, PoolingInstanceAllocator, PoolingInstanceAllocatorConfig,
+    InstanceLimits, PoolConcurrencyLimitError, PoolingInstanceAllocator,
+    PoolingInstanceAllocatorConfig,
 };
 pub use crate::runtime::vm::memory::{Memory, RuntimeLinearMemory, RuntimeMemoryCreator};
 pub use crate::runtime::vm::mmap::Mmap;
diff --git a/crates/wasmtime/src/runtime/vm/cow.rs b/crates/wasmtime/src/runtime/vm/cow.rs
index 057561c6f..c7d70d92b 100644
--- a/crates/wasmtime/src/runtime/vm/cow.rs
+++ b/crates/wasmtime/src/runtime/vm/cow.rs
@@ -470,11 +470,15 @@ impl MemoryImageSlot {
     /// process's memory on Linux. Up to that much memory will be `memset` to
     /// zero where the rest of it will be reset or released with `madvise`.
     #[allow(dead_code)] // ignore warnings as this is only used in some cfgs
-    pub(crate) fn clear_and_remain_ready(&mut self, keep_resident: usize) -> Result<()> {
+    pub(crate) fn clear_and_remain_ready(
+        &mut self,
+        keep_resident: usize,
+        decommit: impl FnMut(*mut u8, usize),
+    ) -> Result<()> {
         assert!(self.dirty);
 
         unsafe {
-            self.reset_all_memory_contents(keep_resident)?;
+            self.reset_all_memory_contents(keep_resident, decommit)?;
         }
 
         self.dirty = false;
@@ -482,7 +486,11 @@ impl MemoryImageSlot {
     }
 
     #[allow(dead_code)] // ignore warnings as this is only used in some cfgs
-    unsafe fn reset_all_memory_contents(&mut self, keep_resident: usize) -> Result<()> {
+    unsafe fn reset_all_memory_contents(
+        &mut self,
+        keep_resident: usize,
+        decommit: impl FnMut(*mut u8, usize),
+    ) -> Result<()> {
         match vm::decommit_behavior() {
             DecommitBehavior::Zero => {
                 // If we're not on Linux then there's no generic platform way to
@@ -494,13 +502,18 @@ impl MemoryImageSlot {
                 self.reset_with_anon_memory()
             }
             DecommitBehavior::RestoreOriginalMapping => {
-                self.reset_with_original_mapping(keep_resident)
+                self.reset_with_original_mapping(keep_resident, decommit);
+                Ok(())
             }
         }
     }
 
     #[allow(dead_code)] // ignore warnings as this is only used in some cfgs
-    unsafe fn reset_with_original_mapping(&mut self, keep_resident: usize) -> Result<()> {
+    unsafe fn reset_with_original_mapping(
+        &mut self,
+        keep_resident: usize,
+        mut decommit: impl FnMut(*mut u8, usize),
+    ) {
         match &self.image {
             Some(image) => {
                 assert!(self.accessible >= image.linear_memory_offset + image.len);
@@ -544,7 +557,11 @@ impl MemoryImageSlot {
                     ptr::write_bytes(self.base.as_ptr(), 0u8, image.linear_memory_offset);
 
                     // This is madvise (2)
-                    self.restore_original_mapping(image.linear_memory_offset, image.len)?;
+                    self.restore_original_mapping(
+                        image.linear_memory_offset,
+                        image.len,
+                        &mut decommit,
+                    );
 
                     // This is memset (3)
                     ptr::write_bytes(self.base.as_ptr().add(image_end), 0u8, remaining_memset);
@@ -553,7 +570,8 @@ impl MemoryImageSlot {
                     self.restore_original_mapping(
                         image_end + remaining_memset,
                         mem_after_image - remaining_memset,
-                    )?;
+                        &mut decommit,
+                    );
                 } else {
                     // If the image starts after the `keep_resident` threshold
                     // then we memset the start of linear memory and then use
@@ -578,7 +596,11 @@ impl MemoryImageSlot {
                     ptr::write_bytes(self.base.as_ptr(), 0u8, keep_resident);
 
                     // This is madvise (2)
-                    self.restore_original_mapping(keep_resident, self.accessible - keep_resident)?;
+                    self.restore_original_mapping(
+                        keep_resident,
+                        self.accessible - keep_resident,
+                        decommit,
+                    );
                 }
             }
 
@@ -588,26 +610,32 @@ impl MemoryImageSlot {
             None => {
                 let size_to_memset = keep_resident.min(self.accessible);
                 ptr::write_bytes(self.base.as_ptr(), 0u8, size_to_memset);
-                self.restore_original_mapping(size_to_memset, self.accessible - size_to_memset)?;
+                self.restore_original_mapping(
+                    size_to_memset,
+                    self.accessible - size_to_memset,
+                    decommit,
+                );
             }
         }
-
-        Ok(())
     }
 
     #[allow(dead_code)] // ignore warnings as this is only used in some cfgs
-    unsafe fn restore_original_mapping(&self, base: usize, len: usize) -> Result<()> {
+    unsafe fn restore_original_mapping(
+        &self,
+        base: usize,
+        len: usize,
+        mut decommit: impl FnMut(*mut u8, usize),
+    ) {
         assert!(base + len <= self.accessible);
         if len == 0 {
-            return Ok(());
+            return;
         }
 
         assert_eq!(
             vm::decommit_behavior(),
             DecommitBehavior::RestoreOriginalMapping
         );
-        vm::decommit_pages(self.base.as_ptr().add(base), len).err2anyhow()?;
-        Ok(())
+        decommit(self.base.as_ptr().add(base), len);
     }
 
     fn set_protection(&self, range: Range<usize>, readwrite: bool) -> Result<()> {
@@ -699,11 +727,11 @@ impl Drop for MemoryImageSlot {
 
 #[cfg(all(test, target_os = "linux", not(miri)))]
 mod test {
-    use std::sync::Arc;
-
     use super::{MemoryImage, MemoryImageSlot, MemoryImageSource, MemoryPlan, MemoryStyle};
     use crate::runtime::vm::mmap::Mmap;
+    use crate::runtime::vm::sys::vm::decommit_pages;
     use anyhow::Result;
+    use std::sync::Arc;
     use wasmtime_environ::Memory;
 
     fn create_memfd_with_data(offset: usize, data: &[u8]) -> Result<MemoryImage> {
@@ -764,7 +792,9 @@ mod test {
         assert_eq!(0, slice[131071]);
         // instantiate again; we should see zeroes, even as the
         // reuse-anon-mmap-opt kicks in
-        memfd.clear_and_remain_ready(0).unwrap();
+        memfd
+            .clear_and_remain_ready(0, |ptr, len| unsafe { decommit_pages(ptr, len).unwrap() })
+            .unwrap();
         assert!(!memfd.is_dirty());
         memfd.instantiate(64 << 10, None, &plan).unwrap();
         let slice = unsafe { mmap.slice(0..65536) };
@@ -790,31 +820,41 @@ mod test {
         assert_eq!(&[1, 2, 3, 4], &slice[4096..4100]);
         slice[4096] = 5;
         // Clear and re-instantiate same image
-        memfd.clear_and_remain_ready(0).unwrap();
+        memfd
+            .clear_and_remain_ready(0, |ptr, len| unsafe { decommit_pages(ptr, len).unwrap() })
+            .unwrap();
         memfd.instantiate(64 << 10, Some(&image), &plan).unwrap();
         let slice = unsafe { mmap.slice_mut(0..65536) };
         // Should not see mutation from above
         assert_eq!(&[1, 2, 3, 4], &slice[4096..4100]);
         // Clear and re-instantiate no image
-        memfd.clear_and_remain_ready(0).unwrap();
+        memfd
+            .clear_and_remain_ready(0, |ptr, len| unsafe { decommit_pages(ptr, len).unwrap() })
+            .unwrap();
         memfd.instantiate(64 << 10, None, &plan).unwrap();
         assert!(!memfd.has_image());
         let slice = unsafe { mmap.slice_mut(0..65536) };
         assert_eq!(&[0, 0, 0, 0], &slice[4096..4100]);
         // Clear and re-instantiate image again
-        memfd.clear_and_remain_ready(0).unwrap();
+        memfd
+            .clear_and_remain_ready(0, |ptr, len| unsafe { decommit_pages(ptr, len).unwrap() })
+            .unwrap();
         memfd.instantiate(64 << 10, Some(&image), &plan).unwrap();
         let slice = unsafe { mmap.slice_mut(0..65536) };
         assert_eq!(&[1, 2, 3, 4], &slice[4096..4100]);
         // Create another image with different data.
         let image2 = Arc::new(create_memfd_with_data(4096, &[10, 11, 12, 13]).unwrap());
-        memfd.clear_and_remain_ready(0).unwrap();
+        memfd
+            .clear_and_remain_ready(0, |ptr, len| unsafe { decommit_pages(ptr, len).unwrap() })
+            .unwrap();
         memfd.instantiate(128 << 10, Some(&image2), &plan).unwrap();
         let slice = unsafe { mmap.slice_mut(0..65536) };
         assert_eq!(&[10, 11, 12, 13], &slice[4096..4100]);
         // Instantiate the original image again; we should notice it's
         // a different image and not reuse the mappings.
-        memfd.clear_and_remain_ready(0).unwrap();
+        memfd
+            .clear_and_remain_ready(0, |ptr, len| unsafe { decommit_pages(ptr, len).unwrap() })
+            .unwrap();
         memfd.instantiate(64 << 10, Some(&image), &plan).unwrap();
         let slice = unsafe { mmap.slice_mut(0..65536) };
         assert_eq!(&[1, 2, 3, 4], &slice[4096..4100]);
@@ -844,7 +884,11 @@ mod test {
                 assert_eq!(&[1, 2, 3, 4], &slice[image_off..][..4]);
                 slice[image_off] = 5;
                 assert_eq!(&[5, 2, 3, 4], &slice[image_off..][..4]);
-                memfd.clear_and_remain_ready(amt_to_memset).unwrap();
+                memfd
+                    .clear_and_remain_ready(amt_to_memset, |ptr, len| unsafe {
+                        decommit_pages(ptr, len).unwrap()
+                    })
+                    .unwrap();
             }
         }
 
@@ -856,7 +900,11 @@ mod test {
                 assert_eq!(chunk[0], 0);
                 chunk[0] = 5;
             }
-            memfd.clear_and_remain_ready(amt_to_memset).unwrap();
+            memfd
+                .clear_and_remain_ready(amt_to_memset, |ptr, len| unsafe {
+                    decommit_pages(ptr, len).unwrap()
+                })
+                .unwrap();
         }
     }
 
@@ -879,7 +927,9 @@ mod test {
         assert_eq!(&[1, 2, 3, 4], &slice[4096..4100]);
         slice[4096] = 5;
         assert_eq!(&[5, 2, 3, 4], &slice[4096..4100]);
-        memfd.clear_and_remain_ready(0).unwrap();
+        memfd
+            .clear_and_remain_ready(0, |ptr, len| unsafe { decommit_pages(ptr, len).unwrap() })
+            .unwrap();
         assert_eq!(&[1, 2, 3, 4], &slice[4096..4100]);
 
         // Re-instantiate make sure it preserves memory. Grow a bit and set data
@@ -890,7 +940,9 @@ mod test {
         assert_eq!(&[0, 0], &slice[initial..initial + 2]);
         slice[initial] = 100;
         assert_eq!(&[100, 0], &slice[initial..initial + 2]);
-        memfd.clear_and_remain_ready(0).unwrap();
+        memfd
+            .clear_and_remain_ready(0, |ptr, len| unsafe { decommit_pages(ptr, len).unwrap() })
+            .unwrap();
 
         // Test that memory is still accessible, but it's been reset
         assert_eq!(&[0, 0], &slice[initial..initial + 2]);
@@ -903,7 +955,9 @@ mod test {
         assert_eq!(&[0, 0], &slice[initial..initial + 2]);
         slice[initial] = 100;
         assert_eq!(&[100, 0], &slice[initial..initial + 2]);
-        memfd.clear_and_remain_ready(0).unwrap();
+        memfd
+            .clear_and_remain_ready(0, |ptr, len| unsafe { decommit_pages(ptr, len).unwrap() })
+            .unwrap();
 
         // Reset the image to none and double-check everything is back to zero
         memfd.instantiate(64 << 10, None, &plan).unwrap();
diff --git a/crates/wasmtime/src/runtime/vm/instance/allocator.rs b/crates/wasmtime/src/runtime/vm/instance/allocator.rs
index 3fa10e930..6309cdf8b 100644
--- a/crates/wasmtime/src/runtime/vm/instance/allocator.rs
+++ b/crates/wasmtime/src/runtime/vm/instance/allocator.rs
@@ -30,7 +30,10 @@ pub use self::on_demand::OnDemandInstanceAllocator;
 #[cfg(feature = "pooling-allocator")]
 mod pooling;
 #[cfg(feature = "pooling-allocator")]
-pub use self::pooling::{InstanceLimits, PoolingInstanceAllocator, PoolingInstanceAllocatorConfig};
+pub use self::pooling::{
+    InstanceLimits, PoolConcurrencyLimitError, PoolingInstanceAllocator,
+    PoolingInstanceAllocatorConfig,
+};
 
 /// Represents a request for a new runtime instance.
 pub struct InstanceAllocationRequest<'a> {
@@ -290,7 +293,7 @@ pub unsafe trait InstanceAllocatorImpl {
     /// The provided stack is required to have been allocated with
     /// `allocate_fiber_stack`.
     #[cfg(feature = "async")]
-    unsafe fn deallocate_fiber_stack(&self, stack: &wasmtime_fiber::FiberStack);
+    unsafe fn deallocate_fiber_stack(&self, stack: wasmtime_fiber::FiberStack);
 
     /// Allocate a GC heap for allocating Wasm GC objects within.
     #[cfg(feature = "gc")]
diff --git a/crates/wasmtime/src/runtime/vm/instance/allocator/on_demand.rs b/crates/wasmtime/src/runtime/vm/instance/allocator/on_demand.rs
index 48fd1009e..c7286c867 100644
--- a/crates/wasmtime/src/runtime/vm/instance/allocator/on_demand.rs
+++ b/crates/wasmtime/src/runtime/vm/instance/allocator/on_demand.rs
@@ -171,8 +171,10 @@ unsafe impl InstanceAllocatorImpl for OnDemandInstanceAllocator {
     }
 
     #[cfg(feature = "async")]
-    unsafe fn deallocate_fiber_stack(&self, _stack: &wasmtime_fiber::FiberStack) {
+    unsafe fn deallocate_fiber_stack(&self, stack: wasmtime_fiber::FiberStack) {
         // The on-demand allocator has no further bookkeeping for fiber stacks
+        // beyond dropping them.
+        let _ = stack;
     }
 
     fn purge_module(&self, _: CompiledModuleId) {}
diff --git a/crates/wasmtime/src/runtime/vm/instance/allocator/pooling.rs b/crates/wasmtime/src/runtime/vm/instance/allocator/pooling.rs
index 6b64fde95..9e3a3caea 100644
--- a/crates/wasmtime/src/runtime/vm/instance/allocator/pooling.rs
+++ b/crates/wasmtime/src/runtime/vm/instance/allocator/pooling.rs
@@ -18,6 +18,7 @@
 //! item is stored in its own separate pool: [`memory_pool`], [`table_pool`],
 //! [`stack_pool`]. See those modules for more details.
 
+mod decommit_queue;
 mod index_allocator;
 mod memory_pool;
 mod table_pool;
@@ -39,6 +40,7 @@ cfg_if::cfg_if! {
     }
 }
 
+use self::decommit_queue::DecommitQueue;
 use self::memory_pool::MemoryPool;
 use self::table_pool::TablePool;
 use super::{
@@ -51,6 +53,9 @@ use crate::runtime::vm::{
     CompiledModuleId, Memory, Table,
 };
 use anyhow::{bail, Result};
+use std::borrow::Cow;
+use std::fmt::Display;
+use std::sync::{Mutex, MutexGuard};
 use std::{
     mem,
     sync::atomic::{AtomicU64, Ordering},
@@ -174,6 +179,11 @@ impl Default for InstanceLimits {
 pub struct PoolingInstanceAllocatorConfig {
     /// See `PoolingAllocatorConfig::max_unused_warm_slots` in `wasmtime`
     pub max_unused_warm_slots: u32,
+    /// The target number of decommits to do per batch. This is not precise, as
+    /// we can queue up decommits at times when we aren't prepared to
+    /// immediately flush them, and so we may go over this target size
+    /// occasionally.
+    pub decommit_batch_size: usize,
     /// The size, in bytes, of async stacks to allocate (not including the guard
     /// page).
     pub stack_size: usize,
@@ -207,6 +217,7 @@ impl Default for PoolingInstanceAllocatorConfig {
     fn default() -> PoolingInstanceAllocatorConfig {
         PoolingInstanceAllocatorConfig {
             max_unused_warm_slots: 100,
+            decommit_batch_size: 1,
             stack_size: 2 << 20,
             limits: InstanceLimits::default(),
             async_stack_zeroing: false,
@@ -219,6 +230,34 @@ impl Default for PoolingInstanceAllocatorConfig {
     }
 }
 
+/// An error returned when the pooling allocator cannot allocate a table,
+/// memory, etc... because the maximum number of concurrent allocations for that
+/// entity has been reached.
+#[derive(Debug)]
+pub struct PoolConcurrencyLimitError {
+    limit: usize,
+    kind: Cow<'static, str>,
+}
+
+impl std::error::Error for PoolConcurrencyLimitError {}
+
+impl Display for PoolConcurrencyLimitError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let limit = self.limit;
+        let kind = &self.kind;
+        write!(f, "maximum concurrent limit of {limit} for {kind} reached")
+    }
+}
+
+impl PoolConcurrencyLimitError {
+    fn new(limit: usize, kind: impl Into<Cow<'static, str>>) -> Self {
+        Self {
+            limit,
+            kind: kind.into(),
+        }
+    }
+}
+
 /// Implements the pooling instance allocator.
 ///
 /// This allocator internally maintains pools of instances, memories, tables,
@@ -228,6 +267,7 @@ impl Default for PoolingInstanceAllocatorConfig {
 /// terminates correctly.
 #[derive(Debug)]
 pub struct PoolingInstanceAllocator {
+    decommit_batch_size: usize,
     limits: InstanceLimits,
 
     // The number of live core module and component instances at any given
@@ -240,6 +280,7 @@ pub struct PoolingInstanceAllocator {
     live_core_instances: AtomicU64,
     live_component_instances: AtomicU64,
 
+    decommit_queue: Mutex<DecommitQueue>,
     memories: MemoryPool,
     tables: TablePool,
 
@@ -250,8 +291,19 @@ pub struct PoolingInstanceAllocator {
     stacks: StackPool,
 }
 
+#[cfg(debug_assertions)]
 impl Drop for PoolingInstanceAllocator {
     fn drop(&mut self) {
+        // NB: when cfg(not(debug_assertions)) it is okay that we don't flush
+        // the queue, as the sub-pools will unmap those ranges anyways, so
+        // there's no point in decommitting them. But we do need to flush the
+        // queue when debug assertions are enabled to make sure that all
+        // entities get returned to their associated sub-pools and we can
+        // differentiate between a leaking slot and an enqueued-for-decommit
+        // slot.
+        let queue = self.decommit_queue.lock().unwrap();
+        self.flush_decommit_queue(queue);
+
         debug_assert_eq!(self.live_component_instances.load(Ordering::Acquire), 0);
         debug_assert_eq!(self.live_core_instances.load(Ordering::Acquire), 0);
 
@@ -270,9 +322,11 @@ impl PoolingInstanceAllocator {
     /// Creates a new pooling instance allocator with the given strategy and limits.
     pub fn new(config: &PoolingInstanceAllocatorConfig, tunables: &Tunables) -> Result<Self> {
         Ok(Self {
+            decommit_batch_size: config.decommit_batch_size,
             limits: config.limits,
             live_component_instances: AtomicU64::new(0),
             live_core_instances: AtomicU64::new(0),
+            decommit_queue: Mutex::new(DecommitQueue::default()),
             memories: MemoryPool::new(config, tunables)?,
             tables: TablePool::new(config)?,
             #[cfg(feature = "gc")]
@@ -370,6 +424,64 @@ impl PoolingInstanceAllocator {
             self.limits.component_instance_size
         )
     }
+
+    fn flush_decommit_queue(&self, mut locked_queue: MutexGuard<'_, DecommitQueue>) -> bool {
+        // Take the queue out of the mutex and drop the lock, to minimize
+        // contention.
+        let queue = mem::take(&mut *locked_queue);
+        drop(locked_queue);
+        queue.flush(self)
+    }
+
+    /// Execute `f` and if it returns `Err(PoolConcurrencyLimitError)`, then try
+    /// flushing the decommit queue. If flushing the queue freed up slots, then
+    /// try running `f` again.
+    fn with_flush_and_retry<T>(&self, mut f: impl FnMut() -> Result<T>) -> Result<T> {
+        f().or_else(|e| {
+            if e.is::<PoolConcurrencyLimitError>() {
+                let queue = self.decommit_queue.lock().unwrap();
+                if self.flush_decommit_queue(queue) {
+                    return f();
+                }
+            }
+
+            Err(e)
+        })
+    }
+
+    fn merge_or_flush(&self, mut local_queue: DecommitQueue) {
+        match local_queue.raw_len() {
+            // If we didn't enqueue any regions for decommit, then we must have
+            // either memset the whole entity or eagerly remapped it to zero
+            // because we don't have linux's `madvise(DONTNEED)` semantics. In
+            // either case, the entity slot is ready for reuse immediately.
+            0 => {
+                local_queue.flush(self);
+            }
+
+            // We enqueued at least our batch size of regions for decommit, so
+            // flush the local queue immediately. Don't bother inspecting (or
+            // locking!) the shared queue.
+            n if n >= self.decommit_batch_size => {
+                local_queue.flush(self);
+            }
+
+            // If we enqueued some regions for decommit, but did not reach our
+            // batch size, so we don't want to flush it yet, then merge the
+            // local queue into the shared queue.
+            n => {
+                debug_assert!(n < self.decommit_batch_size);
+                let mut shared_queue = self.decommit_queue.lock().unwrap();
+                shared_queue.append(&mut local_queue);
+                // And if the shared queue now has at least as many regions
+                // enqueued for decommit as our batch size, then we can flush
+                // it.
+                if shared_queue.raw_len() >= self.decommit_batch_size {
+                    self.flush_decommit_queue(shared_queue);
+                }
+            }
+        }
+    }
 }
 
 unsafe impl InstanceAllocatorImpl for PoolingInstanceAllocator {
@@ -450,10 +562,11 @@ unsafe impl InstanceAllocatorImpl for PoolingInstanceAllocator {
         let old_count = self.live_component_instances.fetch_add(1, Ordering::AcqRel);
         if old_count >= u64::from(self.limits.total_component_instances) {
             self.decrement_component_instance_count();
-            bail!(
-                "maximum concurrent component instance limit of {} reached",
-                self.limits.total_component_instances
-            );
+            return Err(PoolConcurrencyLimitError::new(
+                usize::try_from(self.limits.total_component_instances).unwrap(),
+                "component instances",
+            )
+            .into());
         }
         Ok(())
     }
@@ -466,10 +579,11 @@ unsafe impl InstanceAllocatorImpl for PoolingInstanceAllocator {
         let old_count = self.live_core_instances.fetch_add(1, Ordering::AcqRel);
         if old_count >= u64::from(self.limits.total_core_instances) {
             self.decrement_core_instance_count();
-            bail!(
-                "maximum concurrent core instance limit of {} reached",
-                self.limits.total_core_instances
-            );
+            return Err(PoolConcurrencyLimitError::new(
+                usize::try_from(self.limits.total_core_instances).unwrap(),
+                "core instances",
+            )
+            .into());
         }
         Ok(())
     }
@@ -484,7 +598,7 @@ unsafe impl InstanceAllocatorImpl for PoolingInstanceAllocator {
         memory_plan: &MemoryPlan,
         memory_index: DefinedMemoryIndex,
     ) -> Result<(MemoryAllocationIndex, Memory)> {
-        self.memories.allocate(request, memory_plan, memory_index)
+        self.with_flush_and_retry(|| self.memories.allocate(request, memory_plan, memory_index))
     }
 
     unsafe fn deallocate_memory(
@@ -493,7 +607,19 @@ unsafe impl InstanceAllocatorImpl for PoolingInstanceAllocator {
         allocation_index: MemoryAllocationIndex,
         memory: Memory,
     ) {
-        self.memories.deallocate(allocation_index, memory);
+        // Reset the image slot. If there is any error clearing the
+        // image, just drop it here, and let the drop handler for the
+        // slot unmap in a way that retains the address space
+        // reservation.
+        let mut image = memory.unwrap_static_image();
+        let mut queue = DecommitQueue::default();
+        image
+            .clear_and_remain_ready(self.memories.keep_resident, |ptr, len| {
+                queue.push_raw(ptr, len);
+            })
+            .expect("failed to reset memory image");
+        queue.push_memory(allocation_index, image);
+        self.merge_or_flush(queue);
     }
 
     unsafe fn allocate_table(
@@ -502,26 +628,36 @@ unsafe impl InstanceAllocatorImpl for PoolingInstanceAllocator {
         table_plan: &TablePlan,
         _table_index: DefinedTableIndex,
     ) -> Result<(super::TableAllocationIndex, Table)> {
-        self.tables.allocate(request, table_plan)
+        self.with_flush_and_retry(|| self.tables.allocate(request, table_plan))
     }
 
     unsafe fn deallocate_table(
         &self,
         _table_index: DefinedTableIndex,
         allocation_index: TableAllocationIndex,
-        table: Table,
+        mut table: Table,
     ) {
-        self.tables.deallocate(allocation_index, table);
+        let mut queue = DecommitQueue::default();
+        self.tables
+            .reset_table_pages_to_zero(allocation_index, &mut table, |ptr, len| {
+                queue.push_raw(ptr, len);
+            });
+        queue.push_table(allocation_index, table);
+        self.merge_or_flush(queue);
     }
 
     #[cfg(feature = "async")]
     fn allocate_fiber_stack(&self) -> Result<wasmtime_fiber::FiberStack> {
-        self.stacks.allocate()
+        self.with_flush_and_retry(|| self.stacks.allocate())
     }
 
     #[cfg(feature = "async")]
-    unsafe fn deallocate_fiber_stack(&self, stack: &wasmtime_fiber::FiberStack) {
-        self.stacks.deallocate(stack);
+    unsafe fn deallocate_fiber_stack(&self, mut stack: wasmtime_fiber::FiberStack) {
+        let mut queue = DecommitQueue::default();
+        self.stacks
+            .zero_stack(&mut stack, |ptr, len| queue.push_raw(ptr, len));
+        queue.push_stack(stack);
+        self.merge_or_flush(queue);
     }
 
     fn purge_module(&self, module: CompiledModuleId) {
@@ -613,7 +749,7 @@ mod test {
                 assert_eq!(*addr, 0);
                 *addr = 1;
 
-                allocator.deallocate_fiber_stack(&stack);
+                allocator.deallocate_fiber_stack(stack);
             }
         }
 
@@ -647,7 +783,7 @@ mod test {
                 assert_eq!(*addr, i);
                 *addr = i + 1;
 
-                allocator.deallocate_fiber_stack(&stack);
+                allocator.deallocate_fiber_stack(stack);
             }
         }
 
diff --git a/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/decommit_queue.rs b/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/decommit_queue.rs
new file mode 100644
index 000000000..9bc750b5d
--- /dev/null
+++ b/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/decommit_queue.rs
@@ -0,0 +1,194 @@
+//! A queue for batching decommits together.
+//!
+//! We don't immediately decommit a Wasm table/memory/stack/etc... eagerly, but
+//! instead batch them up to be decommitted together. This module implements
+//! that queuing and batching.
+//!
+//! Even when batching is "disabled" we still use this queue. Batching is
+//! disabled by specifying a batch size of one, in which case, this queue will
+//! immediately get flushed everytime we push onto it.
+
+use super::PoolingInstanceAllocator;
+use crate::vm::{MemoryAllocationIndex, MemoryImageSlot, Table, TableAllocationIndex};
+use smallvec::SmallVec;
+
+#[cfg(feature = "async")]
+use wasmtime_fiber::FiberStack;
+
+#[cfg(unix)]
+#[allow(non_camel_case_types)]
+type iovec = libc::iovec;
+
+#[cfg(not(unix))]
+#[allow(non_camel_case_types)]
+struct iovec {
+    iov_base: *mut libc::c_void,
+    iov_len: libc::size_t,
+}
+
+#[repr(transparent)]
+struct IoVec(iovec);
+
+unsafe impl Send for IoVec {}
+unsafe impl Sync for IoVec {}
+
+impl std::fmt::Debug for IoVec {
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        f.debug_struct("IoVec")
+            .field("base", &self.0.iov_base)
+            .field("len", &self.0.iov_len)
+            .finish()
+    }
+}
+
+#[cfg(feature = "async")]
+struct SendSyncStack(FiberStack);
+#[cfg(feature = "async")]
+unsafe impl Send for SendSyncStack {}
+#[cfg(feature = "async")]
+unsafe impl Sync for SendSyncStack {}
+
+#[derive(Default)]
+pub struct DecommitQueue {
+    raw: SmallVec<[IoVec; 2]>,
+    memories: SmallVec<[(MemoryAllocationIndex, MemoryImageSlot); 1]>,
+    tables: SmallVec<[(TableAllocationIndex, Table); 1]>,
+    #[cfg(feature = "async")]
+    stacks: SmallVec<[SendSyncStack; 1]>,
+    //
+    // TODO: GC heaps are not well-integrated with the pooling allocator
+    // yet. Once we better integrate them, we should start (optionally) zeroing
+    // them, and batching that up here.
+    //
+    // #[cfg(feature = "gc")]
+    // pub gc_heaps: SmallVec<[(GcHeapAllocationIndex, Box<dyn GcHeap>); 1]>,
+}
+
+impl std::fmt::Debug for DecommitQueue {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("DecommitQueue")
+            .field("raw", &self.raw)
+            .finish_non_exhaustive()
+    }
+}
+
+impl DecommitQueue {
+    /// Append another queue to this queue.
+    pub fn append(
+        &mut self,
+        Self {
+            raw,
+            memories,
+            tables,
+            #[cfg(feature = "async")]
+            stacks,
+        }: &mut Self,
+    ) {
+        self.raw.append(raw);
+        self.memories.append(memories);
+        self.tables.append(tables);
+        #[cfg(feature = "async")]
+        self.stacks.append(stacks);
+    }
+
+    /// How many raw memory regions are enqueued for decommit?
+    pub fn raw_len(&self) -> usize {
+        self.raw.len()
+    }
+
+    /// Enqueue a region of memory for decommit.
+    ///
+    /// It is the caller's responsibility to push the associated data via
+    /// `self.push_{memory,table,stack}` as appropriate.
+    ///
+    /// # Safety
+    ///
+    /// The enqueued memory regions must be safe to decommit when `flush` is
+    /// called (no other references, not in use, won't be otherwise unmapped,
+    /// etc...).
+    pub unsafe fn push_raw(&mut self, ptr: *mut u8, len: usize) {
+        self.raw.push(IoVec(iovec {
+            iov_base: ptr.cast(),
+            iov_len: len,
+        }));
+    }
+
+    /// Push a memory into the queue.
+    ///
+    /// # Safety
+    ///
+    /// This memory should not be in use, and its decommit regions must have
+    /// already been enqueued via `self.enqueue_raw`.
+    pub unsafe fn push_memory(
+        &mut self,
+        allocation_index: MemoryAllocationIndex,
+        image: MemoryImageSlot,
+    ) {
+        self.memories.push((allocation_index, image));
+    }
+
+    /// Push a table into the queue.
+    ///
+    /// # Safety
+    ///
+    /// This table should not be in use, and its decommit regions must have
+    /// already been enqueued via `self.enqueue_raw`.
+    pub unsafe fn push_table(&mut self, allocation_index: TableAllocationIndex, table: Table) {
+        self.tables.push((allocation_index, table));
+    }
+
+    /// Push a stack into the queue.
+    ///
+    /// # Safety
+    ///
+    /// This stack should not be in use, and its decommit regions must have
+    /// already been enqueued via `self.enqueue_raw`.
+    #[cfg(feature = "async")]
+    pub unsafe fn push_stack(&mut self, stack: FiberStack) {
+        self.stacks.push(SendSyncStack(stack));
+    }
+
+    fn decommit_all_raw(&mut self) {
+        for iovec in self.raw.drain(..) {
+            unsafe {
+                crate::vm::sys::vm::decommit_pages(iovec.0.iov_base.cast(), iovec.0.iov_len)
+                    .expect("failed to decommit pages");
+            }
+        }
+    }
+
+    /// Flush this queue, decommitting all enqueued regions in batch.
+    ///
+    /// Returns `true` if we did any decommits and returned their entities to
+    /// the associated free lists; `false` if the queue was empty.
+    pub fn flush(mut self, pool: &PoolingInstanceAllocator) -> bool {
+        // First, do the raw decommit syscall(s).
+        self.decommit_all_raw();
+
+        // Second, restore the various entities to their associated pools' free
+        // lists. This is safe, and they are ready for reuse, now that their
+        // memory regions have been decommitted.
+        let mut deallocated_any = false;
+        for (allocation_index, image) in self.memories {
+            deallocated_any = true;
+            unsafe {
+                pool.memories.deallocate(allocation_index, image);
+            }
+        }
+        for (allocation_index, table) in self.tables {
+            deallocated_any = true;
+            unsafe {
+                pool.tables.deallocate(allocation_index, table);
+            }
+        }
+        #[cfg(feature = "async")]
+        for stack in self.stacks {
+            deallocated_any = true;
+            unsafe {
+                pool.stacks.deallocate(stack.0);
+            }
+        }
+
+        deallocated_any
+    }
+}
diff --git a/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/gc_heap_pool.rs b/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/gc_heap_pool.rs
index aa2570831..133e3ae3d 100644
--- a/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/gc_heap_pool.rs
+++ b/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/gc_heap_pool.rs
@@ -40,6 +40,7 @@ impl GcHeapPool {
     }
 
     /// Are there zero slots in use right now?
+    #[allow(unused)] // some cfgs don't use this
     pub fn is_empty(&self) -> bool {
         self.index_allocator.is_empty()
     }
diff --git a/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/generic_stack_pool.rs b/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/generic_stack_pool.rs
index 3b8bc9920..659d06345 100644
--- a/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/generic_stack_pool.rs
+++ b/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/generic_stack_pool.rs
@@ -1,6 +1,6 @@
 #![cfg_attr(not(asan), allow(dead_code))]
 
-use crate::runtime::vm::PoolingInstanceAllocatorConfig;
+use crate::{runtime::vm::PoolingInstanceAllocatorConfig, PoolConcurrencyLimitError};
 use anyhow::{bail, Result};
 use std::sync::atomic::{AtomicU64, Ordering};
 
@@ -31,6 +31,7 @@ impl StackPool {
         })
     }
 
+    #[allow(unused)] // some cfgs don't use this
     pub fn is_empty(&self) -> bool {
         self.live_stacks.load(Ordering::Acquire) == 0
     }
@@ -43,10 +44,11 @@ impl StackPool {
         let old_count = self.live_stacks.fetch_add(1, Ordering::AcqRel);
         if old_count >= self.stack_limit {
             self.live_stacks.fetch_sub(1, Ordering::AcqRel);
-            bail!(
-                "maximum concurrent fiber limit of {} reached",
-                self.stack_limit
-            );
+            return Err(PoolConcurrencyLimitError::new(
+                usize::try_from(self.stack_limit).unwrap(),
+                "fibers",
+            )
+            .into());
         }
 
         match wasmtime_fiber::FiberStack::new(self.stack_size) {
@@ -58,9 +60,19 @@ impl StackPool {
         }
     }
 
-    pub unsafe fn deallocate(&self, stack: &wasmtime_fiber::FiberStack) {
+    pub unsafe fn zero_stack(
+        &self,
+        _stack: &mut wasmtime_fiber::FiberStack,
+        _decommit: impl FnMut(*mut u8, usize),
+    ) {
+        // No need to actually zero the stack, since the stack won't ever be
+        // reused on non-unix systems.
+    }
+
+    /// Safety: see the unix implementation.
+    pub unsafe fn deallocate(&self, stack: wasmtime_fiber::FiberStack) {
         self.live_stacks.fetch_sub(1, Ordering::AcqRel);
-        // A no-op as we don't own the fiber stack on Windows.
+        // A no-op as we don't actually own the fiber stack on Windows.
         let _ = stack;
     }
 }
diff --git a/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/index_allocator.rs b/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/index_allocator.rs
index f84fa70a1..72efa6541 100644
--- a/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/index_allocator.rs
+++ b/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/index_allocator.rs
@@ -31,6 +31,7 @@ impl SimpleIndexAllocator {
         SimpleIndexAllocator(ModuleAffinityIndexAllocator::new(capacity, 0))
     }
 
+    #[allow(unused)] // some cfgs don't use this
     pub fn is_empty(&self) -> bool {
         self.0.is_empty()
     }
@@ -175,6 +176,7 @@ impl ModuleAffinityIndexAllocator {
     }
 
     /// Are zero slots in use right now?
+    #[allow(unused)] // some cfgs don't use this
     pub fn is_empty(&self) -> bool {
         let inner = self.0.lock().unwrap();
         !inner
diff --git a/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/memory_pool.rs b/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/memory_pool.rs
index 5e15a536d..f4e08aadf 100644
--- a/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/memory_pool.rs
+++ b/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/memory_pool.rs
@@ -109,29 +109,33 @@ pub struct MemoryPool {
     /// will contain one stripe per available key; otherwise, a single stripe
     /// with an empty key.
     stripes: Vec<Stripe>,
-    // If using a copy-on-write allocation scheme, the slot management. We
-    // dynamically transfer ownership of a slot to a Memory when in
-    // use.
+
+    /// If using a copy-on-write allocation scheme, the slot management. We
+    /// dynamically transfer ownership of a slot to a Memory when in use.
     image_slots: Vec<Mutex<Option<MemoryImageSlot>>>,
+
     /// A description of the various memory sizes used in allocating the
     /// `mapping` slab.
     layout: SlabLayout,
-    // The maximum number of memories that a single core module instance may
-    // use.
-    //
-    // NB: this is needed for validation but does not affect the pool's size.
+
+    /// The maximum number of memories that a single core module instance may
+    /// use.
+    ///
+    /// NB: this is needed for validation but does not affect the pool's size.
     memories_per_instance: usize,
-    // How much linear memory, in bytes, to keep resident after resetting for
-    // use with the next instance. This much memory will be `memset` to zero
-    // when a linear memory is deallocated.
-    //
-    // Memory exceeding this amount in the wasm linear memory will be released
-    // with `madvise` back to the kernel.
-    //
-    // Only applicable on Linux.
-    keep_resident: usize,
-    // Keep track of protection keys handed out to initialized stores; this
-    // allows us to round-robin the assignment of stores to stripes.
+
+    /// How much linear memory, in bytes, to keep resident after resetting for
+    /// use with the next instance. This much memory will be `memset` to zero
+    /// when a linear memory is deallocated.
+    ///
+    /// Memory exceeding this amount in the wasm linear memory will be released
+    /// with `madvise` back to the kernel.
+    ///
+    /// Only applicable on Linux.
+    pub(super) keep_resident: usize,
+
+    /// Keep track of protection keys handed out to initialized stores; this
+    /// allows us to round-robin the assignment of stores to stripes.
     next_available_pkey: AtomicUsize,
 }
 
@@ -295,6 +299,7 @@ impl MemoryPool {
     }
 
     /// Are zero slots in use right now?
+    #[allow(unused)] // some cfgs don't use this
     pub fn is_empty(&self) -> bool {
         self.stripes.iter().all(|s| s.allocator.is_empty())
     }
@@ -323,10 +328,9 @@ impl MemoryPool {
             )
             .map(|slot| StripedAllocationIndex(u32::try_from(slot.index()).unwrap()))
             .ok_or_else(|| {
-                anyhow!(
-                    "maximum concurrent memory limit of {} reached for stripe {}",
+                super::PoolConcurrencyLimitError::new(
                     self.stripes[stripe_index].allocator.len(),
-                    stripe_index
+                    format!("memory stripe {stripe_index}"),
                 )
             })?;
         let allocation_index =
@@ -395,16 +399,15 @@ impl MemoryPool {
     /// The memory must have been previously allocated from this pool and
     /// assigned the given index, must currently be in an allocated state, and
     /// must never be used again.
-    pub unsafe fn deallocate(&self, allocation_index: MemoryAllocationIndex, memory: Memory) {
-        let mut image = memory.unwrap_static_image();
-
-        // Reset the image slot. If there is any error clearing the
-        // image, just drop it here, and let the drop handler for the
-        // slot unmap in a way that retains the address space
-        // reservation.
-        if image.clear_and_remain_ready(self.keep_resident).is_ok() {
-            self.return_memory_image_slot(allocation_index, image);
-        }
+    ///
+    /// The caller must have already called `clear_and_remain_ready` on the
+    /// memory's image and flushed any enqueued decommits for this memory.
+    pub unsafe fn deallocate(
+        &self,
+        allocation_index: MemoryAllocationIndex,
+        image: MemoryImageSlot,
+    ) {
+        self.return_memory_image_slot(allocation_index, image);
 
         let (stripe_index, striped_allocation_index) =
             StripedAllocationIndex::from_unstriped_slot_index(allocation_index, self.stripes.len());
diff --git a/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/table_pool.rs b/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/table_pool.rs
index 223cca4ff..ddbae36ae 100644
--- a/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/table_pool.rs
+++ b/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/table_pool.rs
@@ -2,7 +2,7 @@ use super::{
     index_allocator::{SimpleIndexAllocator, SlotId},
     round_up_to_pow2, TableAllocationIndex,
 };
-use crate::runtime::vm::sys::vm::{commit_pages, decommit_pages};
+use crate::runtime::vm::sys::vm::commit_pages;
 use crate::runtime::vm::{
     InstanceAllocationRequest, Mmap, PoolingInstanceAllocatorConfig, SendSyncPtr, Table,
 };
@@ -95,6 +95,7 @@ impl TablePool {
     }
 
     /// Are there zero slots in use right now?
+    #[allow(unused)] // some cfgs don't use this
     pub fn is_empty(&self) -> bool {
         self.index_allocator.is_empty()
     }
@@ -122,10 +123,7 @@ impl TablePool {
             .alloc()
             .map(|slot| TableAllocationIndex(slot.0))
             .ok_or_else(|| {
-                anyhow!(
-                    "maximum concurrent table limit of {} reached",
-                    self.max_total_tables
-                )
+                super::PoolConcurrencyLimitError::new(self.max_total_tables, "tables")
             })?;
 
         match (|| {
@@ -166,31 +164,45 @@ impl TablePool {
     /// The table must have been previously-allocated by this pool and assigned
     /// the given allocation index, it must currently be allocated, and it must
     /// never be used again.
+    ///
+    /// The caller must have already called `reset_table_pages_to_zero` on the
+    /// memory and flushed any enqueued decommits for this table's memory.
     pub unsafe fn deallocate(&self, allocation_index: TableAllocationIndex, table: Table) {
         assert!(table.is_static());
+        drop(table);
+        self.index_allocator.free(SlotId(allocation_index.0));
+    }
+
+    /// Reset the given table's memory to zero.
+    ///
+    /// Invokes the given `decommit` function for each region of memory that
+    /// needs to be decommitted. It is the caller's responsibility to actually
+    /// perform that decommit before this table is reused.
+    ///
+    /// # Safety
+    ///
+    /// This table must not be in active use, and ready for returning to the
+    /// table pool once it is zeroed and decommitted.
+    pub unsafe fn reset_table_pages_to_zero(
+        &self,
+        allocation_index: TableAllocationIndex,
+        table: &mut Table,
+        mut decommit: impl FnMut(*mut u8, usize),
+    ) {
+        assert!(table.is_static());
+        let base = self.get(allocation_index);
 
         let size = round_up_to_pow2(
             table.size() as usize * mem::size_of::<*mut u8>(),
             self.page_size,
         );
 
-        drop(table);
-
-        let base = self.get(allocation_index);
-        self.reset_table_pages_to_zero(base, size)
-            .expect("failed to decommit table pages");
-
-        self.index_allocator.free(SlotId(allocation_index.0));
-    }
-
-    fn reset_table_pages_to_zero(&self, base: *mut u8, size: usize) -> Result<()> {
+        // `memset` the first `keep_resident` bytes.
         let size_to_memset = size.min(self.keep_resident);
-        unsafe {
-            std::ptr::write_bytes(base, 0, size_to_memset);
-            decommit_pages(base.add(size_to_memset), size - size_to_memset)
-                .context("failed to decommit table page")?;
-        }
-        Ok(())
+        std::ptr::write_bytes(base, 0, size_to_memset);
+
+        // And decommit the the rest of it.
+        decommit(base.add(size_to_memset), size - size_to_memset)
     }
 }
 
diff --git a/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/unix_stack_pool.rs b/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/unix_stack_pool.rs
index f314333a8..43a404b7f 100644
--- a/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/unix_stack_pool.rs
+++ b/crates/wasmtime/src/runtime/vm/instance/allocator/pooling/unix_stack_pool.rs
@@ -4,7 +4,7 @@ use super::{
     index_allocator::{SimpleIndexAllocator, SlotId},
     round_up_to_pow2,
 };
-use crate::runtime::vm::sys::vm::{commit_pages, decommit_pages};
+use crate::runtime::vm::sys::vm::commit_pages;
 use crate::runtime::vm::{Mmap, PoolingInstanceAllocatorConfig};
 use anyhow::{anyhow, bail, Context, Result};
 
@@ -77,6 +77,7 @@ impl StackPool {
     }
 
     /// Are there zero slots in use right now?
+    #[allow(unused)] // some cfgs don't use this
     pub fn is_empty(&self) -> bool {
         self.index_allocator.is_empty()
     }
@@ -90,12 +91,7 @@ impl StackPool {
         let index = self
             .index_allocator
             .alloc()
-            .ok_or_else(|| {
-                anyhow!(
-                    "maximum concurrent fiber limit of {} reached",
-                    self.max_stacks
-                )
-            })?
+            .ok_or_else(|| super::PoolConcurrencyLimitError::new(self.max_stacks, "fibers"))?
             .index();
 
         assert!(index < self.max_stacks);
@@ -118,13 +114,27 @@ impl StackPool {
         }
     }
 
-    /// Deallocate a previously-allocated fiber.
+    /// Zero the given stack, if we are configured to do so.
+    ///
+    /// This will call the given `decommit` function for each region of memory
+    /// that should be decommitted. It is the caller's responsibility to ensure
+    /// that those decommits happen before this stack is reused.
     ///
     /// # Safety
     ///
-    /// The fiber must have been allocated by this pool, must be in an allocated
-    /// state, and must never be used again.
-    pub unsafe fn deallocate(&self, stack: &wasmtime_fiber::FiberStack) {
+    /// The stack must no longer be in use, and ready for returning to the pool
+    /// after it is zeroed and decommitted.
+    pub unsafe fn zero_stack(
+        &self,
+        stack: &mut wasmtime_fiber::FiberStack,
+        mut decommit: impl FnMut(*mut u8, usize),
+    ) {
+        assert!(stack.is_from_raw_parts());
+
+        if !self.async_stack_zeroing {
+            return;
+        }
+
         let top = stack
             .top()
             .expect("fiber stack not allocated from the pool") as usize;
@@ -143,17 +153,6 @@ impl StackPool {
         assert!(start_of_stack >= base && start_of_stack < (base + len));
         assert!((start_of_stack - base) % self.stack_size == 0);
 
-        let index = (start_of_stack - base) / self.stack_size;
-        assert!(index < self.max_stacks);
-
-        if self.async_stack_zeroing {
-            self.zero_stack(bottom_of_stack, stack_size);
-        }
-
-        self.index_allocator.free(SlotId(index as u32));
-    }
-
-    fn zero_stack(&self, bottom: usize, size: usize) {
         // Manually zero the top of the stack to keep the pages resident in
         // memory and avoid future page faults. Use the system to deallocate
         // pages past this. This hopefully strikes a reasonable balance between:
@@ -161,17 +160,51 @@ impl StackPool {
         // * memset for the whole range is probably expensive
         // * madvise for the whole range incurs expensive future page faults
         // * most threads probably don't use most of the stack anyway
-        let size_to_memset = size.min(self.async_stack_keep_resident);
-        unsafe {
-            std::ptr::write_bytes(
-                (bottom + size - size_to_memset) as *mut u8,
-                0,
-                size_to_memset,
-            );
+        let size_to_memset = stack_size.min(self.async_stack_keep_resident);
+        std::ptr::write_bytes(
+            (bottom_of_stack + stack_size - size_to_memset) as *mut u8,
+            0,
+            size_to_memset,
+        );
 
-            // Use the system to reset remaining stack pages to zero.
-            decommit_pages(bottom as _, size - size_to_memset).unwrap();
-        }
+        // Use the system to reset remaining stack pages to zero.
+        decommit(bottom_of_stack as _, stack_size - size_to_memset);
+    }
+
+    /// Deallocate a previously-allocated fiber.
+    ///
+    /// # Safety
+    ///
+    /// The fiber must have been allocated by this pool, must be in an allocated
+    /// state, and must never be used again.
+    ///
+    /// The caller must have already called `zero_stack` on the fiber stack and
+    /// flushed any enqueued decommits for this stack's memory.
+    pub unsafe fn deallocate(&self, stack: wasmtime_fiber::FiberStack) {
+        assert!(stack.is_from_raw_parts());
+
+        let top = stack
+            .top()
+            .expect("fiber stack not allocated from the pool") as usize;
+
+        let base = self.mapping.as_ptr() as usize;
+        let len = self.mapping.len();
+        assert!(
+            top > base && top <= (base + len),
+            "fiber stack top pointer not in range"
+        );
+
+        // Remove the guard page from the size
+        let stack_size = self.stack_size - self.page_size;
+        let bottom_of_stack = top - stack_size;
+        let start_of_stack = bottom_of_stack - self.page_size;
+        assert!(start_of_stack >= base && start_of_stack < (base + len));
+        assert!((start_of_stack - base) % self.stack_size == 0);
+
+        let index = (start_of_stack - base) / self.stack_size;
+        assert!(index < self.max_stacks);
+
+        self.index_allocator.free(SlotId(index as u32));
     }
 }
 
@@ -220,7 +253,7 @@ mod tests {
 
         for stack in stacks {
             unsafe {
-                pool.deallocate(&stack);
+                pool.deallocate(stack);
             }
         }
 
diff --git a/crates/wasmtime/src/runtime/vm/sys/unix/signals.rs b/crates/wasmtime/src/runtime/vm/sys/unix/signals.rs
index 520fc673a..179eea2be 100644
--- a/crates/wasmtime/src/runtime/vm/sys/unix/signals.rs
+++ b/crates/wasmtime/src/runtime/vm/sys/unix/signals.rs
@@ -255,6 +255,12 @@ unsafe fn get_pc_and_fp(cx: *mut libc::c_void, _signum: libc::c_int) -> (*const
                 cx.mc_gpregs.gp_elr as *const u8,
                 cx.mc_gpregs.gp_x[29] as usize,
             )
+        } else if #[cfg(all(target_os = "openbsd", target_arch = "x86_64"))] {
+            let cx = &*(cx as *const libc::ucontext_t);
+            (
+                cx.sc_rip as *const u8,
+                cx.sc_rbp as usize,
+            )
         }
         else {
             compile_error!("unsupported platform");
diff --git a/crates/wasmtime/src/runtime/vm/table.rs b/crates/wasmtime/src/runtime/vm/table.rs
index 8f826f310..f8e9b4124 100644
--- a/crates/wasmtime/src/runtime/vm/table.rs
+++ b/crates/wasmtime/src/runtime/vm/table.rs
@@ -14,7 +14,7 @@ use core::ptr::{self, NonNull};
 use core::slice;
 use sptr::Strict;
 use wasmtime_environ::{
-    TablePlan, Trap, WasmHeapTopType, WasmRefType, FUNCREF_INIT_BIT, FUNCREF_MASK,
+    TablePlan, TableStyle, Trap, WasmHeapTopType, WasmRefType, FUNCREF_INIT_BIT, FUNCREF_MASK,
 };
 
 /// An element going into or coming out of a table.
@@ -113,18 +113,24 @@ impl TaggedFuncRef {
 
     /// Converts the given `ptr`, a valid funcref pointer, into a tagged pointer
     /// by adding in the `FUNCREF_INIT_BIT`.
-    fn from(ptr: *mut VMFuncRef) -> Self {
-        let masked = Strict::map_addr(ptr, |a| a | FUNCREF_INIT_BIT);
-        TaggedFuncRef(masked)
+    fn from(ptr: *mut VMFuncRef, lazy_init: bool) -> Self {
+        if lazy_init {
+            let masked = Strict::map_addr(ptr, |a| a | FUNCREF_INIT_BIT);
+            TaggedFuncRef(masked)
+        } else {
+            TaggedFuncRef(ptr)
+        }
     }
 
     /// Converts a tagged pointer into a `TableElement`, returning `UninitFunc`
     /// for null (not a tagged value) or `FuncRef` for otherwise tagged values.
-    fn into_table_element(self) -> TableElement {
+    fn into_table_element(self, lazy_init: bool) -> TableElement {
         let ptr = self.0;
-        if ptr.is_null() {
+        if lazy_init && ptr.is_null() {
             TableElement::UninitFunc
         } else {
+            // Masking off the tag bit is harmless whether the table uses lazy
+            // init or not.
             let unmasked = Strict::map_addr(ptr, |a| a & FUNCREF_MASK);
             TableElement::FuncRef(unmasked)
         }
@@ -156,6 +162,8 @@ pub struct StaticFuncTable {
     data: SendSyncPtr<[FuncTableElem]>,
     /// The current size of the table.
     size: u32,
+    /// Whether elements of this table are initialized lazily.
+    lazy_init: bool,
 }
 
 pub struct StaticGcRefTable {
@@ -189,6 +197,8 @@ pub struct DynamicFuncTable {
     elements: Vec<FuncTableElem>,
     /// Maximum size that `elements` can grow to.
     maximum: Option<u32>,
+    /// Whether elements of this table are initialized lazily.
+    lazy_init: bool,
 }
 
 pub struct DynamicGcRefTable {
@@ -260,10 +270,12 @@ impl Table {
     /// Create a new dynamic (movable) table instance for the specified table plan.
     pub fn new_dynamic(plan: &TablePlan, store: &mut dyn Store) -> Result<Self> {
         Self::limit_new(plan, store)?;
+        let TableStyle::CallerChecksSignature { lazy_init } = plan.style;
         match wasm_to_table_type(plan.table.wasm_ty) {
             TableElementType::Func => Ok(Self::from(DynamicFuncTable {
                 elements: vec![None; usize::try_from(plan.table.minimum).unwrap()],
                 maximum: plan.table.maximum,
+                lazy_init,
             })),
             TableElementType::GcRef => Ok(Self::from(DynamicGcRefTable {
                 elements: (0..usize::try_from(plan.table.minimum).unwrap())
@@ -307,7 +319,12 @@ impl Table {
                     data.as_non_null().cast::<FuncTableElem>(),
                     cmp::min(len, max),
                 ));
-                Ok(Self::from(StaticFuncTable { data, size }))
+                let TableStyle::CallerChecksSignature { lazy_init } = plan.style;
+                Ok(Self::from(StaticFuncTable {
+                    data,
+                    size,
+                    lazy_init,
+                }))
             }
             TableElementType::GcRef => {
                 let len = {
@@ -405,14 +422,14 @@ impl Table {
     ) -> Result<(), Trap> {
         let dst = usize::try_from(dst).map_err(|_| Trap::TableOutOfBounds)?;
 
-        let elements = self
-            .funcrefs_mut()
+        let (funcrefs, lazy_init) = self.funcrefs_mut();
+        let elements = funcrefs
             .get_mut(dst..)
             .and_then(|s| s.get_mut(..items.len()))
             .ok_or(Trap::TableOutOfBounds)?;
 
         for (item, slot) in items.zip(elements) {
-            *slot = TaggedFuncRef::from(item);
+            *slot = TaggedFuncRef::from(item, lazy_init);
         }
         Ok(())
     }
@@ -464,7 +481,8 @@ impl Table {
 
         match val {
             TableElement::FuncRef(f) => {
-                self.funcrefs_mut()[start..end].fill(TaggedFuncRef::from(f));
+                let (funcrefs, lazy_init) = self.funcrefs_mut();
+                funcrefs[start..end].fill(TaggedFuncRef::from(f, lazy_init));
             }
             TableElement::GcRef(r) => {
                 // Clone the init GC reference into each table slot.
@@ -479,7 +497,8 @@ impl Table {
                 }
             }
             TableElement::UninitFunc => {
-                self.funcrefs_mut()[start..end].fill(TaggedFuncRef::UNINIT);
+                let (funcrefs, _lazy_init) = self.funcrefs_mut();
+                funcrefs[start..end].fill(TaggedFuncRef::UNINIT);
             }
         }
 
@@ -546,7 +565,7 @@ impl Table {
 
         // First resize the storage and then fill with the init value
         match self {
-            Table::Static(StaticTable::Func(StaticFuncTable { data, size })) => {
+            Table::Static(StaticTable::Func(StaticFuncTable { data, size, .. })) => {
                 unsafe {
                     debug_assert!(data.as_ref()[*size as usize..new_size as usize]
                         .iter()
@@ -590,11 +609,13 @@ impl Table {
     pub fn get(&self, gc_store: &mut GcStore, index: u32) -> Option<TableElement> {
         let index = usize::try_from(index).ok()?;
         match self.element_type() {
-            TableElementType::Func => self
-                .funcrefs()
-                .get(index)
-                .copied()
-                .map(|e| e.into_table_element()),
+            TableElementType::Func => {
+                let (funcrefs, lazy_init) = self.funcrefs();
+                funcrefs
+                    .get(index)
+                    .copied()
+                    .map(|e| e.into_table_element(lazy_init))
+            }
             TableElementType::GcRef => self.gc_refs().get(index).map(|r| {
                 let r = r.as_ref().map(|r| gc_store.clone_gc_ref(r));
                 TableElement::GcRef(r)
@@ -616,10 +637,12 @@ impl Table {
         let index = usize::try_from(index).map_err(|_| ())?;
         match elem {
             TableElement::FuncRef(f) => {
-                *self.funcrefs_mut().get_mut(index).ok_or(())? = TaggedFuncRef::from(f);
+                let (funcrefs, lazy_init) = self.funcrefs_mut();
+                *funcrefs.get_mut(index).ok_or(())? = TaggedFuncRef::from(f, lazy_init);
             }
             TableElement::UninitFunc => {
-                *self.funcrefs_mut().get_mut(index).ok_or(())? = TaggedFuncRef::UNINIT;
+                let (funcrefs, _lazy_init) = self.funcrefs_mut();
+                *funcrefs.get_mut(index).ok_or(())? = TaggedFuncRef::UNINIT;
             }
             TableElement::GcRef(e) => {
                 *self.gc_refs_mut().get_mut(index).ok_or(())? = e;
@@ -675,10 +698,12 @@ impl Table {
     /// Return a `VMTableDefinition` for exposing the table to compiled wasm code.
     pub fn vmtable(&mut self) -> VMTableDefinition {
         match self {
-            Table::Static(StaticTable::Func(StaticFuncTable { data, size })) => VMTableDefinition {
-                base: data.as_ptr().cast(),
-                current_elements: *size,
-            },
+            Table::Static(StaticTable::Func(StaticFuncTable { data, size, .. })) => {
+                VMTableDefinition {
+                    base: data.as_ptr().cast(),
+                    current_elements: *size,
+                }
+            }
             Table::Static(StaticTable::GcRef(StaticGcRefTable { data, size })) => {
                 VMTableDefinition {
                     base: data.as_ptr().cast(),
@@ -704,28 +729,52 @@ impl Table {
         self.element_type().matches(val)
     }
 
-    fn funcrefs(&self) -> &[TaggedFuncRef] {
+    fn funcrefs(&self) -> (&[TaggedFuncRef], bool) {
         assert_eq!(self.element_type(), TableElementType::Func);
         match self {
-            Self::Dynamic(DynamicTable::Func(DynamicFuncTable { elements, .. })) => unsafe {
-                slice::from_raw_parts(elements.as_ptr().cast(), elements.len())
-            },
-            Self::Static(StaticTable::Func(StaticFuncTable { data, size })) => unsafe {
-                slice::from_raw_parts(data.as_ptr().cast(), usize::try_from(*size).unwrap())
-            },
+            Self::Dynamic(DynamicTable::Func(DynamicFuncTable {
+                elements,
+                lazy_init,
+                ..
+            })) => (
+                unsafe { slice::from_raw_parts(elements.as_ptr().cast(), elements.len()) },
+                *lazy_init,
+            ),
+            Self::Static(StaticTable::Func(StaticFuncTable {
+                data,
+                size,
+                lazy_init,
+            })) => (
+                unsafe {
+                    slice::from_raw_parts(data.as_ptr().cast(), usize::try_from(*size).unwrap())
+                },
+                *lazy_init,
+            ),
             _ => unreachable!(),
         }
     }
 
-    fn funcrefs_mut(&mut self) -> &mut [TaggedFuncRef] {
+    fn funcrefs_mut(&mut self) -> (&mut [TaggedFuncRef], bool) {
         assert_eq!(self.element_type(), TableElementType::Func);
         match self {
-            Self::Dynamic(DynamicTable::Func(DynamicFuncTable { elements, .. })) => unsafe {
-                slice::from_raw_parts_mut(elements.as_mut_ptr().cast(), elements.len())
-            },
-            Self::Static(StaticTable::Func(StaticFuncTable { data, size })) => unsafe {
-                slice::from_raw_parts_mut(data.as_ptr().cast(), usize::try_from(*size).unwrap())
-            },
+            Self::Dynamic(DynamicTable::Func(DynamicFuncTable {
+                elements,
+                lazy_init,
+                ..
+            })) => (
+                unsafe { slice::from_raw_parts_mut(elements.as_mut_ptr().cast(), elements.len()) },
+                *lazy_init,
+            ),
+            Self::Static(StaticTable::Func(StaticFuncTable {
+                data,
+                size,
+                lazy_init,
+            })) => (
+                unsafe {
+                    slice::from_raw_parts_mut(data.as_ptr().cast(), usize::try_from(*size).unwrap())
+                },
+                *lazy_init,
+            ),
             _ => unreachable!(),
         }
     }
@@ -770,8 +819,9 @@ impl Table {
         match ty {
             TableElementType::Func => {
                 // `funcref` are `Copy`, so just do a mempcy
-                dst_table.funcrefs_mut()[dst_range]
-                    .copy_from_slice(&src_table.funcrefs()[src_range]);
+                let (dst_funcrefs, _lazy_init) = dst_table.funcrefs_mut();
+                let (src_funcrefs, _lazy_init) = src_table.funcrefs();
+                dst_funcrefs[dst_range].copy_from_slice(&src_funcrefs[src_range]);
             }
             TableElementType::GcRef => {
                 assert_eq!(
@@ -810,7 +860,8 @@ impl Table {
         match ty {
             TableElementType::Func => {
                 // `funcref` are `Copy`, so just do a memmove
-                self.funcrefs_mut().copy_within(src_range, dst_range.start);
+                let (funcrefs, _lazy_init) = self.funcrefs_mut();
+                funcrefs.copy_within(src_range, dst_range.start);
             }
             TableElementType::GcRef => {
                 // We need to clone each `externref` while handling overlapping
@@ -842,6 +893,7 @@ impl Default for Table {
         Self::from(StaticFuncTable {
             data: SendSyncPtr::new(NonNull::from(&mut [])),
             size: 0,
+            lazy_init: false,
         })
     }
 }
diff --git a/src/commands/serve.rs b/src/commands/serve.rs
index e50e5228b..37ab377d7 100644
--- a/src/commands/serve.rs
+++ b/src/commands/serve.rs
@@ -10,7 +10,7 @@ use std::{
     },
 };
 use wasmtime::component::{InstancePre, Linker};
-use wasmtime::{Engine, Store, StoreLimits};
+use wasmtime::{Config, Engine, Memory, MemoryType, Store, StoreLimits};
 use wasmtime_wasi::{StreamError, StreamResult, WasiCtx, WasiCtxBuilder, WasiView};
 use wasmtime_wasi_http::io::TokioIo;
 use wasmtime_wasi_http::{
@@ -247,12 +247,10 @@ impl ServeCommand {
     async fn serve(mut self) -> Result<()> {
         use hyper::server::conn::http1;
 
-        // If not explicitly specified then always enable the pooling allocator.
-        // The `wasmtime serve` use case is tailor-made for pooling allocation
-        // and there's not much downside to enabling it.
-        let pooling_allocator_default = Some(true);
-
-        let mut config = self.run.common.config(None, pooling_allocator_default)?;
+        let mut config = self
+            .run
+            .common
+            .config(None, use_pooling_allocator_by_default().unwrap_or(None))?;
         config.wasm_component_model(true);
         config.async_support(true);
 
@@ -555,3 +553,41 @@ impl wasmtime_wasi::HostOutputStream for LogStream {
 impl wasmtime_wasi::Subscribe for LogStream {
     async fn ready(&mut self) {}
 }
+
+/// The pooling allocator is tailor made for the `wasmtime serve` use case, so
+/// try to use it when we can. The main cost of the pooling allocator, however,
+/// is the virtual memory required to run it. Not all systems support the same
+/// amount of virtual memory, for example some aarch64 and riscv64 configuration
+/// only support 39 bits of virtual address space.
+///
+/// The pooling allocator, by default, will request 1000 linear memories each
+/// sized at 6G per linear memory. This is 6T of virtual memory which ends up
+/// being about 42 bits of the address space. This exceeds the 39 bit limit of
+/// some systems, so there the pooling allocator will fail by default.
+///
+/// This function attempts to dynamically determine the hint for the pooling
+/// allocator. This returns `Some(true)` if the pooling allocator should be used
+/// by default, or `None` or an error otherwise.
+///
+/// The method for testing this is to allocate a 0-sized 64-bit linear memory
+/// with a maximum size that's N bits large where we force all memories to be
+/// static. This should attempt to acquire N bits of the virtual address space.
+/// If successful that should mean that the pooling allocator is OK to use, but
+/// if it fails then the pooling allocator is not used and the normal mmap-based
+/// implementation is used instead.
+fn use_pooling_allocator_by_default() -> Result<Option<bool>> {
+    const BITS_TO_TEST: u32 = 42;
+    let mut config = Config::new();
+    config.wasm_memory64(true);
+    config.static_memory_maximum_size(1 << BITS_TO_TEST);
+    let engine = Engine::new(&config)?;
+    let mut store = Store::new(&engine, ());
+    // NB: the maximum size is in wasm pages to take out the 16-bits of wasm
+    // page size here from the maximum size.
+    let ty = MemoryType::new64(0, Some(1 << (BITS_TO_TEST - 16)));
+    if Memory::new(&mut store, ty).is_ok() {
+        Ok(Some(true))
+    } else {
+        Ok(None)
+    }
+}
diff --git a/winch/codegen/src/visitor.rs b/winch/codegen/src/visitor.rs
index 88fa594c3..75c514748 100644
--- a/winch/codegen/src/visitor.rs
+++ b/winch/codegen/src/visitor.rs
@@ -1375,12 +1375,13 @@ where
         // This code assumes that [`Self::emit_lazy_init_funcref`] will
         // push the funcref to the value stack.
         match self.env.translation.module.table_plans[table_index].style {
-            TableStyle::CallerChecksSignature => {
+            TableStyle::CallerChecksSignature { lazy_init: true } => {
                 let funcref_ptr = self.context.stack.peek().map(|v| v.unwrap_reg()).unwrap();
                 self.masm
                     .trapz(funcref_ptr.into(), TrapCode::IndirectCallToNull);
                 self.emit_typecheck_funcref(funcref_ptr.into(), type_index);
             }
+            _ => unimplemented!("Support for eager table init"),
         }
 
         let callee = self.env.funcref(type_index);
@@ -1428,7 +1429,10 @@ where
 
         match heap_type {
             WasmHeapType::Func => match style {
-                TableStyle::CallerChecksSignature => self.emit_lazy_init_funcref(table_index),
+                TableStyle::CallerChecksSignature { lazy_init: true } => {
+                    self.emit_lazy_init_funcref(table_index)
+                }
+                _ => unimplemented!("Support for eager table init"),
             },
             t => unimplemented!("Support for WasmHeapType: {t}"),
         }
@@ -1501,7 +1505,7 @@ where
         let plan = self.env.table_plan(table_index);
         match plan.table.wasm_ty.heap_type {
             WasmHeapType::Func => match plan.style {
-                TableStyle::CallerChecksSignature => {
+                TableStyle::CallerChecksSignature { lazy_init: true } => {
                     let value = self.context.pop_to_reg(self.masm, None);
                     let index = self.context.pop_to_reg(self.masm, None);
                     let base = self.context.any_gpr(self.masm);
@@ -1521,6 +1525,7 @@ where
                     self.context.free_reg(index);
                     self.context.free_reg(base);
                 }
+                _ => unimplemented!("Support for eager table init"),
             },
             ty => unimplemented!("Support for WasmHeapType: {ty}"),
         };


Test patch:
diff --git a/crates/test-programs/src/bin/cli_large_env.rs b/crates/test-programs/src/bin/cli_large_env.rs
new file mode 100644
index 000000000..7d4a69b5d
--- /dev/null
+++ b/crates/test-programs/src/bin/cli_large_env.rs
@@ -0,0 +1,5 @@
+fn main() {
+    for (k, v) in std::env::vars() {
+        println!("{k}={v}");
+    }
+}
diff --git a/crates/test-programs/src/bin/preview2_file_read_write.rs b/crates/test-programs/src/bin/preview2_file_read_write.rs
index 05a32a65b..8c39d6de5 100644
--- a/crates/test-programs/src/bin/preview2_file_read_write.rs
+++ b/crates/test-programs/src/bin/preview2_file_read_write.rs
@@ -22,7 +22,31 @@ fn main() {
     let stream = file.read_via_stream(0).unwrap();
     let contents = stream.blocking_read(100).unwrap();
     assert_eq!(contents, b"\0\0\0\0\0Hello, World!");
-    drop((stream, file));
+    drop(stream);
+
+    // Test that file read streams behave like other read streams.
+    let mut buf = Vec::new();
+    let stream = file.read_via_stream(0).unwrap();
+    let ready = stream.subscribe();
+    loop {
+        ready.block();
+
+        match stream.read(0) {
+            Ok(chunk) => assert!(chunk.is_empty()),
+            Err(wasi::io::streams::StreamError::Closed) => break,
+            Err(e) => panic!("Failed checking stream state: {e:?}"),
+        }
+
+        match stream.read(4) {
+            Ok(chunk) => buf.extend(chunk),
+            Err(wasi::io::streams::StreamError::Closed) => break,
+            Err(e) => panic!("Failed reading stream: {e:?}"),
+        }
+    }
+    assert_eq!(buf, b"\0\0\0\0\0Hello, World!");
+    drop(ready);
+    drop(stream);
+    drop(file);
 
     dir.unlink_file_at(filename).unwrap();
 }
diff --git a/tests/all/cli_tests.rs b/tests/all/cli_tests.rs
index 89fb70bd4..7f76087d4 100644
--- a/tests/all/cli_tests.rs
+++ b/tests/all/cli_tests.rs
@@ -1818,6 +1818,30 @@ mod test_programs {
         );
         Ok(())
     }
+
+    #[test]
+    fn cli_large_env() -> Result<()> {
+        for wasm in [CLI_LARGE_ENV, CLI_LARGE_ENV_COMPONENT] {
+            println!("run {wasm:?}");
+            let mut cmd = get_wasmtime_command()?;
+            cmd.arg("run").arg("-Sinherit-env").arg(wasm);
+
+            let debug_cmd = format!("{cmd:?}");
+            for i in 0..512 {
+                let var = format!("KEY{i}");
+                let val = (0..1024).map(|_| 'x').collect::<String>();
+                cmd.env(&var, &val);
+            }
+            let output = cmd.output()?;
+            if !output.status.success() {
+                bail!(
+                    "Failed to execute wasmtime with: {debug_cmd}\n{}",
+                    String::from_utf8_lossy(&output.stderr)
+                );
+            }
+        }
+        Ok(())
+    }
 }
 
 #[test]
diff --git a/tests/all/pooling_allocator.rs b/tests/all/pooling_allocator.rs
index b33910209..eceb330fc 100644
--- a/tests/all/pooling_allocator.rs
+++ b/tests/all/pooling_allocator.rs
@@ -430,13 +430,7 @@ fn total_core_instances_limit() -> Result<()> {
 
         match Instance::new(&mut store, &module, &[]) {
             Ok(_) => panic!("instantiation should fail"),
-            Err(e) => assert_eq!(
-                e.to_string(),
-                format!(
-                    "maximum concurrent core instance limit of {} reached",
-                    INSTANCE_LIMIT
-                )
-            ),
+            Err(e) => assert!(e.is::<PoolConcurrencyLimitError>()),
         }
     }
 
@@ -650,7 +644,7 @@ fn instance_too_large() -> Result<()> {
     let engine = Engine::new(&config)?;
     let expected = if cfg!(feature = "wmemcheck") {
         "\
-instance allocation for this module requires 336 bytes which exceeds the \
+        instance allocation for this module requires 336 bytes which exceeds the \
 configured maximum of 16 bytes; breakdown of allocation requirement:
 
  * 76.19% - 256 bytes - instance state management
@@ -867,13 +861,7 @@ fn total_component_instances_limit() -> Result<()> {
 
     match linker.instantiate(&mut store, &component) {
         Ok(_) => panic!("should have hit component instance limit"),
-        Err(e) => assert_eq!(
-            e.to_string(),
-            format!(
-                "maximum concurrent component instance limit of {} reached",
-                TOTAL_COMPONENT_INSTANCES
-            ),
-        ),
+        Err(e) => assert!(e.is::<PoolConcurrencyLimitError>()),
     }
 
     drop(store);
@@ -929,10 +917,7 @@ fn total_tables_limit() -> Result<()> {
 
     match linker.instantiate(&mut store, &module) {
         Ok(_) => panic!("should have hit table limit"),
-        Err(e) => assert_eq!(
-            e.to_string(),
-            format!("maximum concurrent table limit of {} reached", TOTAL_TABLES),
-        ),
+        Err(e) => assert!(e.is::<PoolConcurrencyLimitError>()),
     }
 
     drop(store);
@@ -1006,10 +991,7 @@ async fn total_stacks_limit() -> Result<()> {
     let mut store3 = Store::new(&engine, ());
     match linker.instantiate_async(&mut store3, &module).await {
         Ok(_) => panic!("should have hit stack limit"),
-        Err(e) => assert_eq!(
-            e.to_string(),
-            format!("maximum concurrent fiber limit of {} reached", TOTAL_STACKS),
-        ),
+        Err(e) => assert!(e.is::<PoolConcurrencyLimitError>()),
     }
 
     // Finish the futures and return their Wasm stacks to the pool.
@@ -1185,13 +1167,7 @@ fn total_memories_limit() -> Result<()> {
 
     match linker.instantiate(&mut store, &module) {
         Ok(_) => panic!("should have hit memory limit"),
-        Err(e) => assert_eq!(
-            e.to_string(),
-            format!(
-                "maximum concurrent memory limit of {} reached for stripe 0",
-                TOTAL_MEMORIES
-            ),
-        ),
+        Err(e) => assert!(e.is::<PoolConcurrencyLimitError>()),
     }
 
     drop(store);
@@ -1202,3 +1178,41 @@ fn total_memories_limit() -> Result<()> {
 
     Ok(())
 }
+
+#[test]
+#[cfg_attr(miri, ignore)]
+fn decommit_batching() -> Result<()> {
+    for (capacity, batch_size) in [
+        // A reasonable batch size.
+        (10, 5),
+        // Batch sizes of zero and one should effectively disable batching.
+        (10, 1),
+        (10, 0),
+        // A bigger batch size than capacity, which forces the allocation path
+        // to flush the decommit queue.
+        (10, 99),
+    ] {
+        let mut pool = crate::small_pool_config();
+        pool.total_memories(capacity)
+            .total_core_instances(capacity)
+            .decommit_batch_size(batch_size)
+            .memory_protection_keys(MpkEnabled::Disable);
+        let mut config = Config::new();
+        config.allocation_strategy(InstanceAllocationStrategy::Pooling(pool));
+
+        let engine = Engine::new(&config)?;
+        let linker = Linker::new(&engine);
+        let module = Module::new(&engine, "(module (memory 1 1))")?;
+
+        // Just make sure that we can instantiate all slots a few times and the
+        // pooling allocator must be flushing the decommit queue as necessary.
+        for _ in 0..3 {
+            let mut store = Store::new(&engine, ());
+            for _ in 0..capacity {
+                linker.instantiate(&mut store, &module)?;
+            }
+        }
+    }
+
+    Ok(())
+}
diff --git a/tests/disas/typed-funcrefs-eager-init.wat b/tests/disas/typed-funcrefs-eager-init.wat
new file mode 100644
index 000000000..dc044749c
--- /dev/null
+++ b/tests/disas/typed-funcrefs-eager-init.wat
@@ -0,0 +1,211 @@
+;;! target = "x86_64"
+;;! test = "optimize"
+;;! flags = [ "-Wfunction-references=y", "-Otable-lazy-init=n" ]
+
+;; This test is meant to simulate how typed funcrefs in a table may be
+;; used for ICs (inline caches) in a Wasm module compiled from a dynamic
+;; language. In native JIT engines, IC chains have head pointers that
+;; are raw code pointers and IC-using code can call each with a few ops
+;; (load pointer, call indirect). We'd like similar efficiency by
+;; storing funcrefs for the first IC in each chain in a typed-funcref
+;; table.
+
+(module
+  (type $ic-stub (func (param i32 i32 i32 i32) (result i32)))
+
+  ;; This syntax declares a table that is exactly 100 elements, whose
+  ;; elements are nullable function references, and whose default
+  ;; value is `null`.
+  (table $ic-sites 100 100 (ref null $ic-stub))
+
+  (func $ic1 (param i32 i32 i32 i32) (result i32)
+        local.get 0)
+
+  ;; A function which uses ICs through `table.get` plus `call_ref`
+  (func $call-ics-with-call-ref (param i32 i32 i32 i32) (result i32)
+        (local $sum i32)
+
+        ;; IC callsite index 1 (arbitrary).
+        local.get 0
+        local.get 1
+        local.get 2
+        local.get 3
+        i32.const 1
+        table.get $ic-sites
+        call_ref $ic-stub
+        local.get $sum
+        i32.add
+        local.set $sum
+
+        ;; IC callsite index 2 (arbitrary).
+        local.get 0
+        local.get 1
+        local.get 2
+        local.get 3
+        i32.const 2
+        table.get $ic-sites
+        call_ref $ic-stub
+        local.get $sum
+        i32.add
+        local.set $sum
+
+        local.get $sum)
+
+  ;; Same as the above function, but uses `call_indirect` rather than
+  ;; `call_ref`.
+  (func $call-ics-with-call-indirect (param i32 i32 i32 i32) (result i32)
+        (local $sum i32)
+
+        ;; IC callsite index 1 (arbitrary).
+        local.get 0
+        local.get 1
+        local.get 2
+        local.get 3
+        i32.const 1
+        call_indirect $ic-sites (type $ic-stub)
+        local.get $sum
+        i32.add
+        local.set $sum
+
+        ;; IC callsite index 2 (arbitrary).
+        local.get 0
+        local.get 1
+        local.get 2
+        local.get 3
+        i32.const 2
+        call_indirect $ic-sites (type $ic-stub)
+        local.get $sum
+        i32.add
+        local.set $sum
+
+        local.get $sum)
+
+  (global $ic-site0 (mut (ref $ic-stub)) (ref.func $ic1))
+  (global $ic-site1 (mut (ref $ic-stub)) (ref.func $ic1))
+
+  ;; Sort of similar to the previous two functions, but uses globals instead of
+  ;; tables to store ICs. Mostly just here for comparison in terms of codegen.
+  (func $call-ics-with-global-get (param i32 i32 i32 i32) (result i32)
+        (local $sum i32)
+
+        ;; IC callsite index 1 (arbitrary).
+        local.get 0
+        local.get 1
+        local.get 2
+        local.get 3
+        global.get $ic-site0
+        call_ref $ic-stub
+        local.get $sum
+        i32.add
+        local.set $sum
+
+        ;; IC callsite index 2 (arbitrary).
+        local.get 0
+        local.get 1
+        local.get 2
+        local.get 3
+        global.get $ic-site1
+        call_ref $ic-stub
+        local.get $sum
+        i32.add
+        local.set $sum
+
+        local.get $sum)
+)
+
+;; function u0:0(i64 vmctx, i64, i32, i32, i32, i32) -> i32 fast {
+;;     gv0 = vmctx
+;;     gv1 = load.i64 notrap aligned readonly gv0+8
+;;     gv2 = load.i64 notrap aligned gv1
+;;     stack_limit = gv2
+;;
+;;                                 block0(v0: i64, v1: i64, v2: i32, v3: i32, v4: i32, v5: i32):
+;; @0039                               jump block1
+;;
+;;                                 block1:
+;; @0039                               return v2
+;; }
+;;
+;; function u0:1(i64 vmctx, i64, i32, i32, i32, i32) -> i32 fast {
+;;     gv0 = vmctx
+;;     gv1 = load.i64 notrap aligned readonly gv0+8
+;;     gv2 = load.i64 notrap aligned gv1
+;;     gv3 = vmctx
+;;     gv4 = load.i64 notrap aligned readonly gv3+88
+;;     sig0 = (i64 vmctx, i64, i32, i32, i32, i32) -> i32 fast
+;;     stack_limit = gv2
+;;
+;;                                 block0(v0: i64, v1: i64, v2: i32, v3: i32, v4: i32, v5: i32):
+;; @0048                               v12 = load.i64 notrap aligned readonly v0+88
+;;                                     v48 = iconst.i64 8
+;; @0048                               v14 = iadd v12, v48  ; v48 = 8
+;; @0048                               v17 = load.i64 table_oob aligned table v14
+;; @004a                               v18 = load.i64 null_reference aligned readonly v17+16
+;; @004a                               v19 = load.i64 notrap aligned readonly v17+32
+;; @004a                               v20 = call_indirect sig0, v18(v19, v0, v2, v3, v4, v5)
+;;                                     v56 = iconst.i64 16
+;; @005b                               v28 = iadd v12, v56  ; v56 = 16
+;; @005b                               v31 = load.i64 table_oob aligned table v28
+;; @005d                               v32 = load.i64 null_reference aligned readonly v31+16
+;; @005d                               v33 = load.i64 notrap aligned readonly v31+32
+;; @005d                               v34 = call_indirect sig0, v32(v33, v0, v2, v3, v4, v5)
+;; @0066                               jump block1
+;;
+;;                                 block1:
+;; @0061                               v35 = iadd.i32 v34, v20
+;; @0066                               return v35
+;; }
+;;
+;; function u0:2(i64 vmctx, i64, i32, i32, i32, i32) -> i32 fast {
+;;     gv0 = vmctx
+;;     gv1 = load.i64 notrap aligned readonly gv0+8
+;;     gv2 = load.i64 notrap aligned gv1
+;;     gv3 = vmctx
+;;     gv4 = load.i64 notrap aligned readonly gv3+88
+;;     sig0 = (i64 vmctx, i64, i32, i32, i32, i32) -> i32 fast
+;;     stack_limit = gv2
+;;
+;;                                 block0(v0: i64, v1: i64, v2: i32, v3: i32, v4: i32, v5: i32):
+;; @0075                               v12 = load.i64 notrap aligned readonly v0+88
+;;                                     v48 = iconst.i64 8
+;; @0075                               v14 = iadd v12, v48  ; v48 = 8
+;; @0075                               v17 = load.i64 table_oob aligned table v14
+;; @0075                               v18 = load.i64 icall_null aligned readonly v17+16
+;; @0075                               v19 = load.i64 notrap aligned readonly v17+32
+;; @0075                               v20 = call_indirect sig0, v18(v19, v0, v2, v3, v4, v5)
+;;                                     v56 = iconst.i64 16
+;; @0087                               v28 = iadd v12, v56  ; v56 = 16
+;; @0087                               v31 = load.i64 table_oob aligned table v28
+;; @0087                               v32 = load.i64 icall_null aligned readonly v31+16
+;; @0087                               v33 = load.i64 notrap aligned readonly v31+32
+;; @0087                               v34 = call_indirect sig0, v32(v33, v0, v2, v3, v4, v5)
+;; @0091                               jump block1
+;;
+;;                                 block1:
+;; @008c                               v35 = iadd.i32 v34, v20
+;; @0091                               return v35
+;; }
+;;
+;; function u0:3(i64 vmctx, i64, i32, i32, i32, i32) -> i32 fast {
+;;     gv0 = vmctx
+;;     gv1 = load.i64 notrap aligned readonly gv0+8
+;;     gv2 = load.i64 notrap aligned gv1
+;;     gv3 = vmctx
+;;     sig0 = (i64 vmctx, i64, i32, i32, i32, i32) -> i32 fast
+;;     stack_limit = gv2
+;;
+;;                                 block0(v0: i64, v1: i64, v2: i32, v3: i32, v4: i32, v5: i32):
+;; @009e                               v9 = load.i64 notrap aligned table v0+112
+;; @00a0                               v10 = load.i64 null_reference aligned readonly v9+16
+;; @00a0                               v11 = load.i64 notrap aligned readonly v9+32
+;; @00a0                               v12 = call_indirect sig0, v10(v11, v0, v2, v3, v4, v5)
+;; @00af                               v15 = load.i64 notrap aligned table v0+128
+;; @00b1                               v16 = load.i64 null_reference aligned readonly v15+16
+;; @00b1                               v17 = load.i64 notrap aligned readonly v15+32
+;; @00b1                               v18 = call_indirect sig0, v16(v17, v0, v2, v3, v4, v5)
+;; @00ba                               jump block1
+;;
+;;                                 block1:
+;; @00b5                               v19 = iadd.i32 v18, v12
+;; @00ba                               return v19
+;; }
diff --git a/tests/disas/typed-funcrefs.wat b/tests/disas/typed-funcrefs.wat
index a9567d539..7877c32dc 100644
--- a/tests/disas/typed-funcrefs.wat
+++ b/tests/disas/typed-funcrefs.wat
@@ -1,6 +1,6 @@
 ;;! target = "x86_64"
 ;;! test = "optimize"
-;;! flags = [ "-Wfunction-references=y" ]
+;;! flags = [ "-Wfunction-references=y", "-Otable-lazy-init=y" ]
 
 ;; This test is meant to simulate how typed funcrefs in a table may be
 ;; used for ICs (inline caches) in a Wasm module compiled from a dynamic