diff --git a/Cargo.lock b/Cargo.lock
index 73dbdecd48..6b76f57d61 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3450,6 +3450,7 @@ dependencies = [
 name = "wasmtime-runtime"
 version = "0.24.0"
 dependencies = [
+ "anyhow",
  "backtrace",
  "cc",
  "cfg-if 1.0.0",
diff --git a/crates/cache/src/lib.rs b/crates/cache/src/lib.rs
index fc50ff4dad..9fe69c5baf 100644
--- a/crates/cache/src/lib.rs
+++ b/crates/cache/src/lib.rs
@@ -43,7 +43,7 @@ impl<'config> ModuleCacheEntry<'config> {
     }
 
     /// Gets cached data if state matches, otherwise calls the `compute`.
-    pub fn get_data<T, U, E>(&self, state: T, compute: impl Fn(T) -> Result<U, E>) -> Result<U, E>
+    pub fn get_data<T, U, E>(&self, state: T, compute: fn(T) -> Result<U, E>) -> Result<U, E>
     where
         T: Hash,
         U: Serialize + for<'a> Deserialize<'a>,
diff --git a/crates/cache/src/tests.rs b/crates/cache/src/tests.rs
index 857a1f0ab7..4362aaba22 100644
--- a/crates/cache/src/tests.rs
+++ b/crates/cache/src/tests.rs
@@ -65,68 +65,28 @@ fn test_write_read_cache() {
     let entry1 = ModuleCacheEntry::from_inner(ModuleCacheEntryInner::new(compiler1, &cache_config));
     let entry2 = ModuleCacheEntry::from_inner(ModuleCacheEntryInner::new(compiler2, &cache_config));
 
-    entry1
-        .get_data(1, |_| -> Result<i32, ()> { Ok(100) })
-        .unwrap();
-    entry1
-        .get_data(1, |_| -> Result<i32, ()> { panic!() })
-        .unwrap();
+    entry1.get_data::<_, i32, i32>(1, |_| Ok(100)).unwrap();
+    entry1.get_data::<_, i32, i32>(1, |_| panic!()).unwrap();
 
-    entry1
-        .get_data(2, |_| -> Result<i32, ()> { Ok(100) })
-        .unwrap();
-    entry1
-        .get_data(1, |_| -> Result<i32, ()> { panic!() })
-        .unwrap();
-    entry1
-        .get_data(2, |_| -> Result<i32, ()> { panic!() })
-        .unwrap();
+    entry1.get_data::<_, i32, i32>(2, |_| Ok(100)).unwrap();
+    entry1.get_data::<_, i32, i32>(1, |_| panic!()).unwrap();
+    entry1.get_data::<_, i32, i32>(2, |_| panic!()).unwrap();
 
-    entry1
-        .get_data(3, |_| -> Result<i32, ()> { Ok(100) })
-        .unwrap();
-    entry1
-        .get_data(1, |_| -> Result<i32, ()> { panic!() })
-        .unwrap();
-    entry1
-        .get_data(2, |_| -> Result<i32, ()> { panic!() })
-        .unwrap();
-    entry1
-        .get_data(3, |_| -> Result<i32, ()> { panic!() })
-        .unwrap();
+    entry1.get_data::<_, i32, i32>(3, |_| Ok(100)).unwrap();
+    entry1.get_data::<_, i32, i32>(1, |_| panic!()).unwrap();
+    entry1.get_data::<_, i32, i32>(2, |_| panic!()).unwrap();
+    entry1.get_data::<_, i32, i32>(3, |_| panic!()).unwrap();
 
-    entry1
-        .get_data(4, |_| -> Result<i32, ()> { Ok(100) })
-        .unwrap();
-    entry1
-        .get_data(1, |_| -> Result<i32, ()> { panic!() })
-        .unwrap();
-    entry1
-        .get_data(2, |_| -> Result<i32, ()> { panic!() })
-        .unwrap();
-    entry1
-        .get_data(3, |_| -> Result<i32, ()> { panic!() })
-        .unwrap();
-    entry1
-        .get_data(4, |_| -> Result<i32, ()> { panic!() })
-        .unwrap();
+    entry1.get_data::<_, i32, i32>(4, |_| Ok(100)).unwrap();
+    entry1.get_data::<_, i32, i32>(1, |_| panic!()).unwrap();
+    entry1.get_data::<_, i32, i32>(2, |_| panic!()).unwrap();
+    entry1.get_data::<_, i32, i32>(3, |_| panic!()).unwrap();
+    entry1.get_data::<_, i32, i32>(4, |_| panic!()).unwrap();
 
-    entry2
-        .get_data(1, |_| -> Result<i32, ()> { Ok(100) })
-        .unwrap();
-    entry1
-        .get_data(1, |_| -> Result<i32, ()> { panic!() })
-        .unwrap();
-    entry1
-        .get_data(2, |_| -> Result<i32, ()> { panic!() })
-        .unwrap();
-    entry1
-        .get_data(3, |_| -> Result<i32, ()> { panic!() })
-        .unwrap();
-    entry1
-        .get_data(4, |_| -> Result<i32, ()> { panic!() })
-        .unwrap();
-    entry2
-        .get_data(1, |_| -> Result<i32, ()> { panic!() })
-        .unwrap();
+    entry2.get_data::<_, i32, i32>(1, |_| Ok(100)).unwrap();
+    entry1.get_data::<_, i32, i32>(1, |_| panic!()).unwrap();
+    entry1.get_data::<_, i32, i32>(2, |_| panic!()).unwrap();
+    entry1.get_data::<_, i32, i32>(3, |_| panic!()).unwrap();
+    entry1.get_data::<_, i32, i32>(4, |_| panic!()).unwrap();
+    entry2.get_data::<_, i32, i32>(1, |_| panic!()).unwrap();
 }
diff --git a/crates/environ/src/module.rs b/crates/environ/src/module.rs
index b70748ec72..ee4ff050dc 100644
--- a/crates/environ/src/module.rs
+++ b/crates/environ/src/module.rs
@@ -1,7 +1,7 @@
 //! Data structures for representing decoded wasm modules.
 
 use crate::tunables::Tunables;
-use crate::{DataInitializer, WASM_MAX_PAGES, WASM_PAGE_SIZE};
+use crate::WASM_MAX_PAGES;
 use cranelift_codegen::ir;
 use cranelift_entity::{EntityRef, PrimaryMap};
 use cranelift_wasm::*;
@@ -92,51 +92,12 @@ pub struct MemoryInitializer {
     pub data: Box<[u8]>,
 }
 
-impl From<DataInitializer<'_>> for MemoryInitializer {
-    fn from(initializer: DataInitializer) -> Self {
-        Self {
-            memory_index: initializer.memory_index,
-            base: initializer.base,
-            offset: initializer.offset,
-            data: initializer.data.into(),
-        }
-    }
-}
-
-/// The type of WebAssembly linear memory initialization.
+/// The type of WebAssembly linear memory initialization to use for a module.
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub enum MemoryInitialization {
-    /// Memory initialization is paged.
-    ///
-    /// To be paged, the following requirements must be met:
-    ///
-    /// * All data segments must reference defined memories.
-    /// * All data segments must not use a global base.
-    /// * All data segments must be in bounds.
-    ///
-    /// Paged initialization is performed by memcopying individual pages to the linear memory.
-    Paged {
-        /// The size of each page stored in the map.
-        /// This is expected to be the host page size.
-        page_size: usize,
-        /// The map of defined memory index to a list of page data.
-        /// The list of page data is sparse, with None representing a zero page.
-        /// The size of the list will be the maximum page written to by a data segment.
-        map: PrimaryMap<DefinedMemoryIndex, Vec<Option<Box<[u8]>>>>,
-    },
-    /// Memory initialization is out of bounds.
-    ///
-    /// To be out of bounds, the following requirements must be met:
-    ///
-    /// * All data segments must reference defined memories.
-    /// * All data segments must not use a global base.
-    /// * At least one data segments was out of bounds.
-    ///
-    /// This can be used to quickly return an error when the module is instantiated.
-    OutOfBounds,
     /// Memory initialization is segmented.
     ///
-    /// To be segmented, at least one of the following requirements must be met:
+    /// Segmented initialization can be used for any module, but it is required if:
     ///
     /// * A data segment referenced an imported memory.
     /// * A data segment uses a global base.
@@ -144,100 +105,131 @@ pub enum MemoryInitialization {
     /// Segmented initialization is performed by processing the complete set of data segments
     /// when the module is instantiated.
     ///
-    /// This ensures that initialization side-effects are observed according to the bulk-memory proposal.
-    Segmented(Box<[MemoryInitializer]>),
+    /// This is the default memory initialization type.
+    Segmented(Vec<MemoryInitializer>),
+    /// Memory initialization is paged.
+    ///
+    /// To be paged, the following requirements must be met:
+    ///
+    /// * All data segments must reference defined memories.
+    /// * All data segments must not use a global base.
+    ///
+    /// Paged initialization is performed by copying (or mapping) entire WebAssembly pages to each linear memory.
+    ///
+    /// The `uffd` feature makes use of this type of memory initialization because it can instruct the kernel
+    /// to back an entire WebAssembly page from an existing set of in-memory pages.
+    ///
+    /// By processing the data segments at module compilation time, the uffd fault handler doesn't have to do
+    /// any work to point the kernel at the right linear memory page to use.
+    Paged {
+        /// The map of defined memory index to a list of initialization pages.
+        /// The list of page data is sparse, with None representing a zero page.
+        /// Each page of initialization data is WebAssembly page-sized (64 KiB).
+        /// The size of the list will be the maximum page written to by a data segment.
+        map: PrimaryMap<DefinedMemoryIndex, Vec<Option<Box<[u8]>>>>,
+        /// Whether or not an out-of-bounds data segment was observed.
+        /// This is used to fail module instantiation after the pages are initialized.
+        out_of_bounds: bool,
+    },
 }
 
 impl MemoryInitialization {
-    /// Creates a new memory initialization for a module and its data initializers.
-    pub fn new(module: &Module, initializers: Vec<DataInitializer>) -> Self {
-        let page_size = region::page::size();
-        let num_defined_memories = module.memory_plans.len() - module.num_imported_memories;
-        let mut out_of_bounds = false;
-        let mut memories = PrimaryMap::with_capacity(num_defined_memories);
+    /// Attempts to convert segmented memory initialization into paged initialization for the given module.
+    ///
+    /// Returns `None` if the initialization cannot be paged or if it is already paged.
+    pub fn to_paged(&self, module: &Module) -> Option<Self> {
+        const WASM_PAGE_SIZE: usize = crate::WASM_PAGE_SIZE as usize;
 
-        for _ in 0..num_defined_memories {
-            memories.push(Vec::new());
-        }
+        match self {
+            Self::Paged { .. } => None,
+            Self::Segmented(initializers) => {
+                let num_defined_memories = module.memory_plans.len() - module.num_imported_memories;
+                let mut out_of_bounds = false;
+                let mut map = PrimaryMap::with_capacity(num_defined_memories);
 
-        for initializer in &initializers {
-            match (
-                module.defined_memory_index(initializer.memory_index),
-                initializer.base.is_some(),
-            ) {
-                (None, _) | (_, true) => {
-                    // If the initializer references an imported memory or uses a global base,
-                    // the complete set of segments will need to be processed at module instantiation
-                    return Self::Segmented(
-                        initializers
-                            .into_iter()
-                            .map(Into::into)
-                            .collect::<Vec<_>>()
-                            .into_boxed_slice(),
-                    );
+                for _ in 0..num_defined_memories {
+                    map.push(Vec::new());
                 }
-                (Some(index), false) => {
-                    if out_of_bounds {
-                        continue;
-                    }
 
-                    // Perform a bounds check on the segment
-                    if (initializer.offset + initializer.data.len())
-                        > ((module.memory_plans[initializer.memory_index].memory.minimum as usize)
-                            * (WASM_PAGE_SIZE as usize))
-                    {
-                        out_of_bounds = true;
-                        continue;
-                    }
-
-                    let pages = &mut memories[index];
-                    let mut page_index = initializer.offset / page_size;
-                    let mut page_offset = initializer.offset % page_size;
-                    let mut data_offset = 0;
-                    let mut data_remaining = initializer.data.len();
-
-                    if data_remaining == 0 {
-                        continue;
-                    }
-
-                    // Copy the initialization data by each page
-                    loop {
-                        if page_index >= pages.len() {
-                            pages.resize(page_index + 1, None);
+                for initializer in initializers {
+                    match (
+                        module.defined_memory_index(initializer.memory_index),
+                        initializer.base.is_some(),
+                    ) {
+                        (None, _) | (_, true) => {
+                            // If the initializer references an imported memory or uses a global base,
+                            // the complete set of segments will need to be processed at module instantiation
+                            return None;
                         }
+                        (Some(index), false) => {
+                            if out_of_bounds {
+                                continue;
+                            }
 
-                        let page = pages[page_index]
-                            .get_or_insert_with(|| vec![0; page_size].into_boxed_slice());
-                        let len = std::cmp::min(data_remaining, page_size - page_offset);
+                            // Perform a bounds check on the segment
+                            // As this segment is referencing a defined memory without a global base, the last byte
+                            // written to by the segment cannot exceed the memory's initial minimum size
+                            if (initializer.offset + initializer.data.len())
+                                > ((module.memory_plans[initializer.memory_index].memory.minimum
+                                    as usize)
+                                    * WASM_PAGE_SIZE)
+                            {
+                                out_of_bounds = true;
+                                continue;
+                            }
 
-                        page[page_offset..page_offset + len]
-                            .copy_from_slice(&initializer.data[data_offset..(data_offset + len)]);
+                            let pages = &mut map[index];
+                            let mut page_index = initializer.offset / WASM_PAGE_SIZE;
+                            let mut page_offset = initializer.offset % WASM_PAGE_SIZE;
+                            let mut data_offset = 0;
+                            let mut data_remaining = initializer.data.len();
 
-                        if len == data_remaining {
-                            break;
+                            if data_remaining == 0 {
+                                continue;
+                            }
+
+                            // Copy the initialization data by each WebAssembly-sized page (64 KiB)
+                            loop {
+                                if page_index >= pages.len() {
+                                    pages.resize(page_index + 1, None);
+                                }
+
+                                let page = pages[page_index].get_or_insert_with(|| {
+                                    vec![0; WASM_PAGE_SIZE].into_boxed_slice()
+                                });
+                                let len =
+                                    std::cmp::min(data_remaining, WASM_PAGE_SIZE - page_offset);
+
+                                page[page_offset..page_offset + len].copy_from_slice(
+                                    &initializer.data[data_offset..(data_offset + len)],
+                                );
+
+                                if len == data_remaining {
+                                    break;
+                                }
+
+                                page_index += 1;
+                                page_offset = 0;
+                                data_offset += len;
+                                data_remaining -= len;
+                            }
                         }
-
-                        page_index += 1;
-                        page_offset = 0;
-                        data_offset += len;
-                        data_remaining -= len;
-                    }
+                    };
                 }
-            };
-        }
 
-        if out_of_bounds {
-            Self::OutOfBounds
-        } else {
-            Self::Paged {
-                page_size,
-                map: memories,
+                Some(Self::Paged { map, out_of_bounds })
             }
         }
     }
 }
 
-/// Implemenation styles for WebAssembly tables.
+impl Default for MemoryInitialization {
+    fn default() -> Self {
+        Self::Segmented(Vec::new())
+    }
+}
+
+/// Implementation styles for WebAssembly tables.
 #[derive(Debug, Clone, Hash, Serialize, Deserialize)]
 pub enum TableStyle {
     /// Signatures are stored in the table and checked in the caller.
@@ -325,7 +317,7 @@ pub struct Module {
     pub table_initializers: Vec<TableInitializer>,
 
     /// WebAssembly linear memory initializer.
-    pub memory_initialization: Option<MemoryInitialization>,
+    pub memory_initialization: MemoryInitialization,
 
     /// WebAssembly passive elements.
     pub passive_elements: Vec<Box<[FuncIndex]>>,
@@ -405,7 +397,7 @@ pub enum Initializer {
         export: String,
     },
 
-    /// A module is being instantiated with previously configured intializers
+    /// A module is being instantiated with previously configured initializers
     /// as arguments.
     Instantiate {
         /// The module that this instance is instantiating.
@@ -417,7 +409,7 @@ pub enum Initializer {
 
     /// A module is being created from a set of compiled artifacts.
     CreateModule {
-        /// The index of the artifact that's being convereted into a module.
+        /// The index of the artifact that's being converted into a module.
         artifact_index: usize,
         /// The list of artifacts that this module value will be inheriting.
         artifacts: Vec<usize>,
diff --git a/crates/environ/src/module_environ.rs b/crates/environ/src/module_environ.rs
index 636fa2893e..2f53bb4af8 100644
--- a/crates/environ/src/module_environ.rs
+++ b/crates/environ/src/module_environ.rs
@@ -1,6 +1,6 @@
 use crate::module::{
-    Initializer, InstanceSignature, MemoryPlan, Module, ModuleSignature, ModuleType, ModuleUpvar,
-    TableInitializer, TablePlan, TypeTables,
+    Initializer, InstanceSignature, MemoryInitialization, MemoryInitializer, MemoryPlan, Module,
+    ModuleSignature, ModuleType, ModuleUpvar, TableInitializer, TablePlan, TypeTables,
 };
 use crate::tunables::Tunables;
 use cranelift_codegen::ir;
@@ -59,9 +59,6 @@ pub struct ModuleTranslation<'data> {
     /// References to the function bodies.
     pub function_body_inputs: PrimaryMap<DefinedFuncIndex, FunctionBodyData<'data>>,
 
-    /// References to the data initializers.
-    pub data_initializers: Vec<DataInitializer<'data>>,
-
     /// DWARF debug information, if enabled, parsed from the module.
     pub debuginfo: DebugInfoData<'data>,
 
@@ -762,9 +759,12 @@ impl<'data> cranelift_wasm::ModuleEnvironment<'data> for ModuleEnvironment<'data
     }
 
     fn reserve_data_initializers(&mut self, num: u32) -> WasmResult<()> {
-        self.result
-            .data_initializers
-            .reserve_exact(usize::try_from(num).unwrap());
+        match &mut self.result.module.memory_initialization {
+            MemoryInitialization::Segmented(initializers) => {
+                initializers.reserve_exact(usize::try_from(num).unwrap())
+            }
+            _ => unreachable!(),
+        }
         Ok(())
     }
 
@@ -775,12 +775,17 @@ impl<'data> cranelift_wasm::ModuleEnvironment<'data> for ModuleEnvironment<'data
         offset: usize,
         data: &'data [u8],
     ) -> WasmResult<()> {
-        self.result.data_initializers.push(DataInitializer {
-            memory_index,
-            base,
-            offset,
-            data,
-        });
+        match &mut self.result.module.memory_initialization {
+            MemoryInitialization::Segmented(initializers) => {
+                initializers.push(MemoryInitializer {
+                    memory_index,
+                    base,
+                    offset,
+                    data: data.into(),
+                });
+            }
+            _ => unreachable!(),
+        }
         Ok(())
     }
 
@@ -1071,18 +1076,3 @@ pub fn translate_signature(mut sig: ir::Signature, pointer_type: ir::Type) -> ir
     sig.params.insert(1, AbiParam::new(pointer_type));
     sig
 }
-
-/// A data initializer for linear memory.
-pub struct DataInitializer<'data> {
-    /// The index of the memory to initialize.
-    pub memory_index: MemoryIndex,
-
-    /// Optionally a globalvar base to initialize at.
-    pub base: Option<GlobalIndex>,
-
-    /// A constant offset to initialize at.
-    pub offset: usize,
-
-    /// The initialization data.
-    pub data: &'data [u8],
-}
diff --git a/crates/jit/src/instantiate.rs b/crates/jit/src/instantiate.rs
index b4875305be..250b837281 100644
--- a/crates/jit/src/instantiate.rs
+++ b/crates/jit/src/instantiate.rs
@@ -21,9 +21,8 @@ use wasmtime_environ::wasm::{
     DefinedFuncIndex, InstanceTypeIndex, ModuleTypeIndex, SignatureIndex, WasmFuncType,
 };
 use wasmtime_environ::{
-    CompileError, DebugInfoData, FunctionAddressMap, InstanceSignature, MemoryInitialization,
-    Module, ModuleEnvironment, ModuleSignature, ModuleTranslation, StackMapInformation,
-    TrapInformation,
+    CompileError, DebugInfoData, FunctionAddressMap, InstanceSignature, Module, ModuleEnvironment,
+    ModuleSignature, ModuleTranslation, StackMapInformation, TrapInformation,
 };
 use wasmtime_profiling::ProfilingAgent;
 use wasmtime_runtime::{GdbJitImageRegistration, InstantiationError, VMFunctionBody, VMTrampoline};
@@ -95,10 +94,14 @@ struct DebugInfo {
 
 impl CompilationArtifacts {
     /// Creates a `CompilationArtifacts` for a singular translated wasm module.
+    ///
+    /// The `use_paged_init` argument controls whether or not an attempt is made to
+    /// organize linear memory initialization data as entire pages or to leave
+    /// the memory initialization data as individual segments.
     pub fn build(
         compiler: &Compiler,
         data: &[u8],
-        validate: impl Fn(&ModuleTranslation) -> Result<(), String> + Sync,
+        use_paged_mem_init: bool,
     ) -> Result<(usize, Vec<CompilationArtifacts>, TypeTables), SetupError> {
         let (main_module, translations, types) = ModuleEnvironment::new(
             compiler.frontend_config(),
@@ -110,8 +113,6 @@ impl CompilationArtifacts {
 
         let list = maybe_parallel!(translations.(into_iter | into_par_iter))
             .map(|mut translation| {
-                validate(&translation).map_err(|e| SetupError::Validate(e))?;
-
                 let Compilation {
                     obj,
                     unwind_info,
@@ -120,14 +121,16 @@ impl CompilationArtifacts {
 
                 let ModuleTranslation {
                     mut module,
-                    data_initializers,
                     debuginfo,
                     has_unparsed_debuginfo,
                     ..
                 } = translation;
 
-                module.memory_initialization =
-                    Some(MemoryInitialization::new(&module, data_initializers));
+                if use_paged_mem_init {
+                    if let Some(init) = module.memory_initialization.to_paged(&module) {
+                        module.memory_initialization = init;
+                    }
+                }
 
                 let obj = obj.write().map_err(|_| {
                     SetupError::Instantiate(InstantiationError::Resource(
diff --git a/crates/obj/src/data_segment.rs b/crates/obj/src/data_segment.rs
index 159922f25b..3e4184eb20 100644
--- a/crates/obj/src/data_segment.rs
+++ b/crates/obj/src/data_segment.rs
@@ -1,12 +1,12 @@
 use anyhow::Result;
 use object::write::{Object, StandardSection, Symbol, SymbolSection};
 use object::{SymbolFlags, SymbolKind, SymbolScope};
-use wasmtime_environ::DataInitializer;
+use wasmtime_environ::MemoryInitializer;
 
 /// Declares data segment symbol
 pub fn declare_data_segment(
     obj: &mut Object,
-    _data_initaliazer: &DataInitializer,
+    _memory_initializer: &MemoryInitializer,
     index: usize,
 ) -> Result<()> {
     let name = format!("_memory_{}", index);
@@ -26,12 +26,12 @@ pub fn declare_data_segment(
 /// Emit segment data and initialization location
 pub fn emit_data_segment(
     obj: &mut Object,
-    data_initaliazer: &DataInitializer,
+    memory_initializer: &MemoryInitializer,
     index: usize,
 ) -> Result<()> {
     let name = format!("_memory_{}", index);
     let symbol_id = obj.symbol_id(name.as_bytes()).unwrap();
     let section_id = obj.section_id(StandardSection::Data);
-    obj.add_symbol_data(symbol_id, section_id, data_initaliazer.data, 1);
+    obj.add_symbol_data(symbol_id, section_id, &memory_initializer.data, 1);
     Ok(())
 }
diff --git a/crates/obj/src/module.rs b/crates/obj/src/module.rs
index 2adf1aa393..4150d1c8b8 100644
--- a/crates/obj/src/module.rs
+++ b/crates/obj/src/module.rs
@@ -7,7 +7,7 @@ use object::write::{Object, Relocation, StandardSection, Symbol, SymbolSection};
 use object::{RelocationEncoding, RelocationKind, SymbolFlags, SymbolKind, SymbolScope};
 use wasmtime_debug::DwarfSection;
 use wasmtime_environ::isa::TargetFrontendConfig;
-use wasmtime_environ::{CompiledFunctions, DataInitializer, Module};
+use wasmtime_environ::{CompiledFunctions, MemoryInitialization, Module};
 
 fn emit_vmcontext_init(
     obj: &mut Object,
@@ -54,24 +54,32 @@ pub fn emit_module(
     target_config: &TargetFrontendConfig,
     compilation: CompiledFunctions,
     dwarf_sections: Vec<DwarfSection>,
-    data_initializers: &[DataInitializer],
 ) -> Result<Object> {
     let mut builder = ObjectBuilder::new(target, module, &compilation);
     builder.set_dwarf_sections(dwarf_sections);
     let mut obj = builder.build()?;
 
     // Append data, table and vmcontext_init code to the object file.
-
-    for (i, initializer) in data_initializers.iter().enumerate() {
-        declare_data_segment(&mut obj, initializer, i)?;
+    match &module.memory_initialization {
+        MemoryInitialization::Segmented(initializers) => {
+            for (i, initializer) in initializers.iter().enumerate() {
+                declare_data_segment(&mut obj, initializer, i)?;
+            }
+        }
+        _ => unimplemented!(),
     }
 
     for i in 0..module.table_plans.len() {
         declare_table(&mut obj, i)?;
     }
 
-    for (i, initializer) in data_initializers.iter().enumerate() {
-        emit_data_segment(&mut obj, initializer, i)?;
+    match &module.memory_initialization {
+        MemoryInitialization::Segmented(initializers) => {
+            for (i, initializer) in initializers.iter().enumerate() {
+                emit_data_segment(&mut obj, initializer, i)?;
+            }
+        }
+        _ => unimplemented!(),
     }
 
     for i in 0..module.table_plans.len() {
diff --git a/crates/runtime/Cargo.toml b/crates/runtime/Cargo.toml
index af73a7d102..31a020fd4e 100644
--- a/crates/runtime/Cargo.toml
+++ b/crates/runtime/Cargo.toml
@@ -25,6 +25,7 @@ backtrace = "0.3.55"
 lazy_static = "1.3.0"
 psm = "0.1.11"
 rand = "0.7.3"
+anyhow = "1.0.38"
 
 [target.'cfg(target_os = "windows")'.dependencies]
 winapi = { version = "0.3.7", features = ["winbase", "memoryapi", "errhandlingapi"] }
diff --git a/crates/runtime/src/instance.rs b/crates/runtime/src/instance.rs
index c90c267f6b..313642a077 100644
--- a/crates/runtime/src/instance.rs
+++ b/crates/runtime/src/instance.rs
@@ -834,12 +834,12 @@ impl Instance {
     ///
     /// Resetting the guard pages is required before growing memory.
     #[cfg(all(feature = "uffd", target_os = "linux"))]
-    pub(crate) fn reset_guard_pages(&self) -> Result<(), String> {
+    pub(crate) fn reset_guard_pages(&self) -> anyhow::Result<()> {
         let mut faults = self.guard_page_faults.borrow_mut();
         for (addr, len, reset) in faults.drain(..) {
             unsafe {
                 if !reset(addr, len) {
-                    return Err("failed to reset previously faulted memory guard page".into());
+                    anyhow::bail!("failed to reset previously faulted memory guard page");
                 }
             }
         }
diff --git a/crates/runtime/src/instance/allocator.rs b/crates/runtime/src/instance/allocator.rs
index aae17c7f4d..c835c14421 100644
--- a/crates/runtime/src/instance/allocator.rs
+++ b/crates/runtime/src/instance/allocator.rs
@@ -9,6 +9,7 @@ use crate::vmcontext::{
     VMGlobalDefinition, VMGlobalImport, VMInterrupts, VMMemoryImport, VMSharedSignatureIndex,
     VMTableImport,
 };
+use anyhow::Result;
 use std::alloc;
 use std::any::Any;
 use std::cell::RefCell;
@@ -23,8 +24,8 @@ use wasmtime_environ::wasm::{
     TableElementType, WasmType,
 };
 use wasmtime_environ::{
-    ir, MemoryInitialization, MemoryInitializer, Module, ModuleTranslation, ModuleType,
-    TableInitializer, VMOffsets,
+    ir, MemoryInitialization, MemoryInitializer, Module, ModuleType, TableInitializer, VMOffsets,
+    WASM_PAGE_SIZE,
 };
 
 mod pooling;
@@ -105,11 +106,9 @@ pub enum FiberStackError {
 ///
 /// This trait is unsafe as it requires knowledge of Wasmtime's runtime internals to implement correctly.
 pub unsafe trait InstanceAllocator: Send + Sync {
-    /// Validates a module translation.
-    ///
-    /// This is used to ensure a module being compiled is supported by the instance allocator.
-    fn validate_module(&self, translation: &ModuleTranslation) -> Result<(), String> {
-        drop(translation);
+    /// Validates that a module is supported by the allocator.
+    fn validate(&self, module: &Module) -> Result<()> {
+        drop(module);
         Ok(())
     }
 
@@ -322,15 +321,14 @@ fn check_init_bounds(instance: &Instance) -> Result<(), InstantiationError> {
     check_table_init_bounds(instance)?;
 
     match &instance.module.memory_initialization {
-        Some(MemoryInitialization::Paged { .. }) | None => {
-            // Bounds were checked at compile-time
+        MemoryInitialization::Paged { out_of_bounds, .. } => {
+            if *out_of_bounds {
+                return Err(InstantiationError::Link(LinkError(
+                    "memory out of bounds: data segment does not fit".into(),
+                )));
+            }
         }
-        Some(MemoryInitialization::OutOfBounds) => {
-            return Err(InstantiationError::Link(LinkError(
-                "memory out of bounds: data segment does not fit".into(),
-            )));
-        }
-        Some(MemoryInitialization::Segmented(initializers)) => {
+        MemoryInitialization::Segmented(initializers) => {
             check_memory_init_bounds(instance, initializers)?;
         }
     }
@@ -355,37 +353,31 @@ fn initialize_instance(
 
     // Initialize the memories
     match &instance.module.memory_initialization {
-        Some(MemoryInitialization::Paged { page_size, map }) => {
+        MemoryInitialization::Paged { map, out_of_bounds } => {
             for (index, pages) in map {
                 let memory = instance.memory(index);
+                let slice =
+                    unsafe { slice::from_raw_parts_mut(memory.base, memory.current_length) };
 
                 for (page_index, page) in pages.iter().enumerate() {
                     if let Some(data) = page {
-                        // Bounds checking should have occurred when the module was compiled
-                        // The data should always be page sized
-                        assert!((page_index * page_size) < memory.current_length);
-                        assert_eq!(data.len(), *page_size);
-
-                        unsafe {
-                            ptr::copy_nonoverlapping(
-                                data.as_ptr(),
-                                memory.base.add(page_index * page_size),
-                                data.len(),
-                            );
-                        }
+                        debug_assert_eq!(data.len(), WASM_PAGE_SIZE as usize);
+                        slice[page_index * WASM_PAGE_SIZE as usize..].copy_from_slice(data);
                     }
                 }
             }
+
+            // Check for out of bound access after initializing the pages to maintain
+            // the expected behavior of the bulk memory spec.
+            if *out_of_bounds {
+                return Err(InstantiationError::Trap(Trap::wasm(
+                    ir::TrapCode::HeapOutOfBounds,
+                )));
+            }
         }
-        Some(MemoryInitialization::OutOfBounds) => {
-            return Err(InstantiationError::Trap(Trap::wasm(
-                ir::TrapCode::HeapOutOfBounds,
-            )))
-        }
-        Some(MemoryInitialization::Segmented(initializers)) => {
+        MemoryInitialization::Segmented(initializers) => {
             initialize_memories(instance, initializers)?;
         }
-        None => {}
     }
 
     Ok(())
@@ -615,10 +607,9 @@ unsafe impl InstanceAllocator for OnDemandInstanceAllocator {
     }
 
     unsafe fn deallocate(&self, handle: &InstanceHandle) {
-        let instance = handle.instance();
-        let layout = instance.alloc_layout();
-        ptr::drop_in_place(instance as *const Instance as *mut Instance);
-        alloc::dealloc(instance as *const Instance as *mut _, layout);
+        let layout = handle.instance().alloc_layout();
+        ptr::drop_in_place(handle.instance);
+        alloc::dealloc(handle.instance.cast(), layout);
     }
 
     fn allocate_fiber_stack(&self) -> Result<*mut u8, FiberStackError> {
diff --git a/crates/runtime/src/instance/allocator/pooling.rs b/crates/runtime/src/instance/allocator/pooling.rs
index c759780ad8..5029d90667 100644
--- a/crates/runtime/src/instance/allocator/pooling.rs
+++ b/crates/runtime/src/instance/allocator/pooling.rs
@@ -12,6 +12,7 @@ use super::{
     InstanceAllocator, InstanceHandle, InstantiationError,
 };
 use crate::{instance::Instance, table::max_table_element_size, Memory, Mmap, Table, VMContext};
+use anyhow::{anyhow, bail, Context, Result};
 use rand::Rng;
 use std::cell::RefCell;
 use std::cmp::min;
@@ -20,7 +21,7 @@ use std::mem;
 use std::sync::{Arc, Mutex};
 use wasmtime_environ::{
     entity::{EntitySet, PrimaryMap},
-    MemoryStyle, Module, ModuleTranslation, Tunables, VMOffsets, WASM_PAGE_SIZE,
+    MemoryStyle, Module, Tunables, VMOffsets, WASM_PAGE_SIZE,
 };
 
 cfg_if::cfg_if! {
@@ -30,10 +31,9 @@ cfg_if::cfg_if! {
     } else if #[cfg(all(feature = "uffd", target_os = "linux"))] {
         mod uffd;
         use uffd as imp;
-        use imp::{PageFaultHandler, reset_guard_page};
+        use imp::PageFaultHandler;
         use super::{check_init_bounds, initialize_tables};
         use wasmtime_environ::MemoryInitialization;
-        use std::sync::atomic::{AtomicBool, Ordering};
     } else if #[cfg(target_os = "linux")] {
         mod linux;
         use linux as imp;
@@ -105,73 +105,81 @@ pub struct ModuleLimits {
 }
 
 impl ModuleLimits {
-    fn validate_module(&self, module: &Module) -> Result<(), String> {
+    fn validate(&self, module: &Module) -> Result<()> {
         if module.num_imported_funcs > self.imported_functions as usize {
-            return Err(format!(
+            bail!(
                 "imported function count of {} exceeds the limit of {}",
-                module.num_imported_funcs, self.imported_functions
-            ));
+                module.num_imported_funcs,
+                self.imported_functions
+            );
         }
 
         if module.num_imported_tables > self.imported_tables as usize {
-            return Err(format!(
+            bail!(
                 "imported tables count of {} exceeds the limit of {}",
-                module.num_imported_tables, self.imported_tables
-            ));
+                module.num_imported_tables,
+                self.imported_tables
+            );
         }
 
         if module.num_imported_memories > self.imported_memories as usize {
-            return Err(format!(
+            bail!(
                 "imported memories count of {} exceeds the limit of {}",
-                module.num_imported_memories, self.imported_memories
-            ));
+                module.num_imported_memories,
+                self.imported_memories
+            );
         }
 
         if module.num_imported_globals > self.imported_globals as usize {
-            return Err(format!(
+            bail!(
                 "imported globals count of {} exceeds the limit of {}",
-                module.num_imported_globals, self.imported_globals
-            ));
+                module.num_imported_globals,
+                self.imported_globals
+            );
         }
 
         if module.types.len() > self.types as usize {
-            return Err(format!(
+            bail!(
                 "defined types count of {} exceeds the limit of {}",
                 module.types.len(),
                 self.types
-            ));
+            );
         }
 
         let functions = module.functions.len() - module.num_imported_funcs;
         if functions > self.functions as usize {
-            return Err(format!(
+            bail!(
                 "defined functions count of {} exceeds the limit of {}",
-                functions, self.functions
-            ));
+                functions,
+                self.functions
+            );
         }
 
         let tables = module.table_plans.len() - module.num_imported_tables;
         if tables > self.tables as usize {
-            return Err(format!(
+            bail!(
                 "defined tables count of {} exceeds the limit of {}",
-                tables, self.tables
-            ));
+                tables,
+                self.tables
+            );
         }
 
         let memories = module.memory_plans.len() - module.num_imported_memories;
         if memories > self.memories as usize {
-            return Err(format!(
+            bail!(
                 "defined memories count of {} exceeds the limit of {}",
-                memories, self.memories
-            ));
+                memories,
+                self.memories
+            );
         }
 
         let globals = module.globals.len() - module.num_imported_globals;
         if globals > self.globals as usize {
-            return Err(format!(
+            bail!(
                 "defined globals count of {} exceeds the limit of {}",
-                globals, self.globals
-            ));
+                globals,
+                self.globals
+            );
         }
 
         for (i, plan) in module.table_plans.values().as_slice()[module.num_imported_tables..]
@@ -179,10 +187,12 @@ impl ModuleLimits {
             .enumerate()
         {
             if plan.table.minimum > self.table_elements {
-                return Err(format!(
+                bail!(
                     "table index {} has a minimum element size of {} which exceeds the limit of {}",
-                    i, plan.table.minimum, self.table_elements
-                ));
+                    i,
+                    plan.table.minimum,
+                    self.table_elements
+                );
             }
         }
 
@@ -191,17 +201,19 @@ impl ModuleLimits {
             .enumerate()
         {
             if plan.memory.minimum > self.memory_pages {
-                return Err(format!(
+                bail!(
                     "memory index {} has a minimum page size of {} which exceeds the limit of {}",
-                    i, plan.memory.minimum, self.memory_pages
-                ));
+                    i,
+                    plan.memory.minimum,
+                    self.memory_pages
+                );
             }
 
             if let MemoryStyle::Dynamic = plan.style {
-                return Err(format!(
+                bail!(
                     "memory index {} has an unsupported dynamic memory plan style",
                     i,
-                ));
+                );
             }
         }
 
@@ -353,7 +365,7 @@ struct InstancePool {
 }
 
 impl InstancePool {
-    fn new(module_limits: &ModuleLimits, instance_limits: &InstanceLimits) -> Result<Self, String> {
+    fn new(module_limits: &ModuleLimits, instance_limits: &InstanceLimits) -> Result<Self> {
         let page_size = region::page::size();
 
         // Calculate the maximum size of an Instance structure given the limits
@@ -373,7 +385,7 @@ impl InstancePool {
         let instance_size = round_up_to_pow2(
             mem::size_of::<Instance>()
                 .checked_add(offsets.size_of_vmctx() as usize)
-                .ok_or_else(|| "instance size exceeds addressable memory".to_string())?,
+                .ok_or_else(|| anyhow!("instance size exceeds addressable memory"))?,
             page_size,
         );
 
@@ -381,7 +393,7 @@ impl InstancePool {
 
         let allocation_size = instance_size
             .checked_mul(max_instances)
-            .ok_or_else(|| "total size of instance data exceeds addressable memory".to_string())?;
+            .ok_or_else(|| anyhow!("total size of instance data exceeds addressable memory"))?;
 
         let pool = Self {
             mapping: create_memory_map(allocation_size, allocation_size)?,
@@ -527,7 +539,7 @@ impl InstancePool {
         #[cfg(all(feature = "uffd", target_os = "linux"))]
         instance
             .reset_guard_pages()
-            .map_err(InstantiationError::Resource)?;
+            .map_err(|e| InstantiationError::Resource(e.to_string()))?;
 
         instance.memories.clear();
 
@@ -610,9 +622,9 @@ struct MemoryPool {
 }
 
 impl MemoryPool {
-    fn new(module_limits: &ModuleLimits, instance_limits: &InstanceLimits) -> Result<Self, String> {
+    fn new(module_limits: &ModuleLimits, instance_limits: &InstanceLimits) -> Result<Self> {
         let memory_size = usize::try_from(instance_limits.memory_reservation_size)
-            .map_err(|_| "memory reservation size exceeds addressable memory".to_string())?;
+            .map_err(|_| anyhow!("memory reservation size exceeds addressable memory"))?;
 
         debug_assert!(
             memory_size % region::page::size() == 0,
@@ -627,7 +639,7 @@ impl MemoryPool {
             .checked_mul(max_memories)
             .and_then(|c| c.checked_mul(max_instances))
             .ok_or_else(|| {
-                "total size of memory reservation exceeds addressable memory".to_string()
+                anyhow!("total size of memory reservation exceeds addressable memory")
             })?;
 
         Ok(Self {
@@ -670,13 +682,13 @@ struct TablePool {
 }
 
 impl TablePool {
-    fn new(module_limits: &ModuleLimits, instance_limits: &InstanceLimits) -> Result<Self, String> {
+    fn new(module_limits: &ModuleLimits, instance_limits: &InstanceLimits) -> Result<Self> {
         let page_size = region::page::size();
 
         let table_size = round_up_to_pow2(
             max_table_element_size()
                 .checked_mul(module_limits.table_elements as usize)
-                .ok_or_else(|| "table size exceeds addressable memory".to_string())?,
+                .ok_or_else(|| anyhow!("table size exceeds addressable memory"))?,
             page_size,
         );
 
@@ -686,9 +698,7 @@ impl TablePool {
         let allocation_size = table_size
             .checked_mul(max_tables)
             .and_then(|c| c.checked_mul(max_instances))
-            .ok_or_else(|| {
-                "total size of instance tables exceeds addressable memory".to_string()
-            })?;
+            .ok_or_else(|| anyhow!("total size of instance tables exceeds addressable memory"))?;
 
         Ok(Self {
             mapping: create_memory_map(0, allocation_size)?,
@@ -733,12 +743,10 @@ struct StackPool {
     max_instances: usize,
     page_size: usize,
     free_list: Mutex<Vec<usize>>,
-    #[cfg(all(feature = "uffd", target_os = "linux"))]
-    faulted_guard_pages: Arc<[AtomicBool]>,
 }
 
 impl StackPool {
-    fn new(instance_limits: &InstanceLimits, stack_size: usize) -> Result<Self, String> {
+    fn new(instance_limits: &InstanceLimits, stack_size: usize) -> Result<Self> {
         let page_size = region::page::size();
 
         // On Windows, don't allocate any fiber stacks as native fibers are always used
@@ -748,26 +756,33 @@ impl StackPool {
         } else {
             round_up_to_pow2(stack_size, page_size)
                 .checked_add(page_size)
-                .ok_or_else(|| "stack size exceeds addressable memory".to_string())?
+                .ok_or_else(|| anyhow!("stack size exceeds addressable memory"))?
         };
 
         let max_instances = instance_limits.count as usize;
 
-        let allocation_size = stack_size.checked_mul(max_instances).ok_or_else(|| {
-            "total size of execution stacks exceeds addressable memory".to_string()
-        })?;
+        let allocation_size = stack_size
+            .checked_mul(max_instances)
+            .ok_or_else(|| anyhow!("total size of execution stacks exceeds addressable memory"))?;
+
+        let mapping = create_memory_map(allocation_size, allocation_size)?;
+
+        // Set up the stack guard pages
+        unsafe {
+            for i in 0..max_instances {
+                // Make the stack guard page inaccessible
+                let bottom_of_stack = mapping.as_mut_ptr().add(i * stack_size);
+                region::protect(bottom_of_stack, page_size, region::Protection::NONE)
+                    .context("failed to protect stack guard page")?;
+            }
+        }
 
         Ok(Self {
-            mapping: create_memory_map(0, allocation_size)?,
+            mapping,
             stack_size,
             max_instances,
             page_size,
             free_list: Mutex::new((0..max_instances).collect()),
-            #[cfg(all(feature = "uffd", target_os = "linux"))]
-            faulted_guard_pages: std::iter::repeat_with(|| false.into())
-                .take(max_instances)
-                .collect::<Vec<_>>()
-                .into(),
         })
     }
 
@@ -789,37 +804,8 @@ impl StackPool {
         debug_assert!(index < self.max_instances);
 
         unsafe {
-            // Remove the guard page from the size
-            let size_without_guard = self.stack_size - self.page_size;
-
-            let bottom_of_stack = self
-                .mapping
-                .as_mut_ptr()
-                .add((index * self.stack_size) + self.page_size);
-
-            cfg_if::cfg_if! {
-                if #[cfg(all(feature = "uffd", target_os = "linux"))] {
-                    // Check to see if a guard page needs to be reset
-                    if self.faulted_guard_pages[index].swap(false, Ordering::SeqCst) {
-                        if !reset_guard_page(bottom_of_stack.sub(self.page_size), self.page_size) {
-                            return Err(FiberStackError::Resource(
-                                "failed to reset stack guard page".into(),
-                            ));
-                        }
-                    }
-
-                } else {
-                    // Make the stack accessible (excluding the guard page)
-                    if !make_accessible(bottom_of_stack, size_without_guard) {
-                        return Err(FiberStackError::Resource(
-                            "failed to make instance memory accessible".into(),
-                        ));
-                    }
-                }
-            }
-
-            // The top of the stack should be returned
-            Ok(bottom_of_stack.add(size_without_guard))
+            // The top (end) of the stack should be returned
+            Ok(self.mapping.as_mut_ptr().add((index + 1) * self.stack_size))
         }
     }
 
@@ -872,9 +858,9 @@ impl PoolingInstanceAllocator {
         module_limits: ModuleLimits,
         mut instance_limits: InstanceLimits,
         stack_size: usize,
-    ) -> Result<Self, String> {
+    ) -> Result<Self> {
         if instance_limits.count == 0 {
-            return Err("the instance count limit cannot be zero".into());
+            bail!("the instance count limit cannot be zero");
         }
 
         // Round the memory reservation size to the nearest Wasm page size
@@ -890,28 +876,28 @@ impl PoolingInstanceAllocator {
 
         // The maximum module memory page count cannot exceed 65536 pages
         if module_limits.memory_pages > 0x10000 {
-            return Err(format!(
+            bail!(
                 "module memory page limit of {} exceeds the maximum of 65536",
                 module_limits.memory_pages
-            ));
+            );
         }
 
         // The maximum module memory page count cannot exceed the memory reservation size
         if (module_limits.memory_pages * WASM_PAGE_SIZE) as u64
             > instance_limits.memory_reservation_size
         {
-            return Err(format!(
+            bail!(
                 "module memory page limit of {} pages exeeds the memory reservation size limit of {} bytes",
                 module_limits.memory_pages,
                 instance_limits.memory_reservation_size
-            ));
+            );
         }
 
         let instances = InstancePool::new(&module_limits, &instance_limits)?;
         let stacks = StackPool::new(&instance_limits, stack_size)?;
 
         #[cfg(all(feature = "uffd", target_os = "linux"))]
-        let _fault_handler = PageFaultHandler::new(&instances, &stacks)?;
+        let _fault_handler = PageFaultHandler::new(&instances)?;
 
         Ok(Self {
             strategy,
@@ -937,8 +923,8 @@ impl Drop for PoolingInstanceAllocator {
 }
 
 unsafe impl InstanceAllocator for PoolingInstanceAllocator {
-    fn validate_module(&self, translation: &ModuleTranslation) -> Result<(), String> {
-        self.module_limits.validate_module(&translation.module)
+    fn validate(&self, module: &Module) -> Result<()> {
+        self.module_limits.validate(module)
     }
 
     fn adjust_tunables(&self, tunables: &mut Tunables) {
@@ -976,8 +962,8 @@ unsafe impl InstanceAllocator for PoolingInstanceAllocator {
 
         cfg_if::cfg_if! {
             if #[cfg(all(feature = "uffd", target_os = "linux"))] {
-                match instance.module.memory_initialization {
-                    Some(MemoryInitialization::Paged{ .. }) => {
+                match &instance.module.memory_initialization {
+                    MemoryInitialization::Paged{ out_of_bounds, .. } => {
                         if !is_bulk_memory {
                             check_init_bounds(instance)?;
                         }
@@ -985,7 +971,15 @@ unsafe impl InstanceAllocator for PoolingInstanceAllocator {
                         // Initialize the tables
                         initialize_tables(instance)?;
 
-                        // Don't initialize the memory; the fault handler will fill the pages when accessed
+                        // Don't initialize the memory; the fault handler will back the pages when accessed
+
+                        // If there was an out of bounds access observed in initialization, return a trap
+                        if *out_of_bounds {
+                            return Err(InstantiationError::Trap(crate::traphandlers::Trap::wasm(
+                                wasmtime_environ::ir::TrapCode::HeapOutOfBounds,
+                            )));
+                        }
+
                         Ok(())
                     },
                     _ => initialize_instance(instance, is_bulk_memory)
@@ -1030,11 +1024,11 @@ mod test {
         let mut module = Module::default();
 
         module.functions.push(SignatureIndex::new(0));
-        assert_eq!(limits.validate_module(&module), Ok(()));
+        assert!(limits.validate(&module).is_ok());
 
         module.num_imported_funcs = 1;
         assert_eq!(
-            limits.validate_module(&module),
+            limits.validate(&module).map_err(|e| e.to_string()),
             Err("imported function count of 1 exceeds the limit of 0".into())
         );
     }
@@ -1058,11 +1052,11 @@ mod test {
             },
         });
 
-        assert_eq!(limits.validate_module(&module), Ok(()));
+        assert!(limits.validate(&module).is_ok());
 
         module.num_imported_tables = 1;
         assert_eq!(
-            limits.validate_module(&module),
+            limits.validate(&module).map_err(|e| e.to_string()),
             Err("imported tables count of 1 exceeds the limit of 0".into())
         );
     }
@@ -1086,11 +1080,11 @@ mod test {
             offset_guard_size: 0,
         });
 
-        assert_eq!(limits.validate_module(&module), Ok(()));
+        assert!(limits.validate(&module).is_ok());
 
         module.num_imported_memories = 1;
         assert_eq!(
-            limits.validate_module(&module),
+            limits.validate(&module).map_err(|e| e.to_string()),
             Err("imported memories count of 1 exceeds the limit of 0".into())
         );
     }
@@ -1111,11 +1105,11 @@ mod test {
             initializer: GlobalInit::I32Const(0),
         });
 
-        assert_eq!(limits.validate_module(&module), Ok(()));
+        assert!(limits.validate(&module).is_ok());
 
         module.num_imported_globals = 1;
         assert_eq!(
-            limits.validate_module(&module),
+            limits.validate(&module).map_err(|e| e.to_string()),
             Err("imported globals count of 1 exceeds the limit of 0".into())
         );
     }
@@ -1128,13 +1122,13 @@ mod test {
         };
 
         let mut module = Module::default();
-        assert_eq!(limits.validate_module(&module), Ok(()));
+        assert!(limits.validate(&module).is_ok());
 
         module
             .types
             .push(ModuleType::Function(SignatureIndex::new(0)));
         assert_eq!(
-            limits.validate_module(&module),
+            limits.validate(&module).map_err(|e| e.to_string()),
             Err("defined types count of 1 exceeds the limit of 0".into())
         );
     }
@@ -1147,11 +1141,11 @@ mod test {
         };
 
         let mut module = Module::default();
-        assert_eq!(limits.validate_module(&module), Ok(()));
+        assert!(limits.validate(&module).is_ok());
 
         module.functions.push(SignatureIndex::new(0));
         assert_eq!(
-            limits.validate_module(&module),
+            limits.validate(&module).map_err(|e| e.to_string()),
             Err("defined functions count of 1 exceeds the limit of 0".into())
         );
     }
@@ -1164,7 +1158,7 @@ mod test {
         };
 
         let mut module = Module::default();
-        assert_eq!(limits.validate_module(&module), Ok(()));
+        assert!(limits.validate(&module).is_ok());
 
         module.table_plans.push(TablePlan {
             style: TableStyle::CallerChecksSignature,
@@ -1176,7 +1170,7 @@ mod test {
             },
         });
         assert_eq!(
-            limits.validate_module(&module),
+            limits.validate(&module).map_err(|e| e.to_string()),
             Err("defined tables count of 1 exceeds the limit of 0".into())
         );
     }
@@ -1189,7 +1183,7 @@ mod test {
         };
 
         let mut module = Module::default();
-        assert_eq!(limits.validate_module(&module), Ok(()));
+        assert!(limits.validate(&module).is_ok());
 
         module.memory_plans.push(MemoryPlan {
             style: MemoryStyle::Static { bound: 0 },
@@ -1201,7 +1195,7 @@ mod test {
             offset_guard_size: 0,
         });
         assert_eq!(
-            limits.validate_module(&module),
+            limits.validate(&module).map_err(|e| e.to_string()),
             Err("defined memories count of 1 exceeds the limit of 0".into())
         );
     }
@@ -1214,7 +1208,7 @@ mod test {
         };
 
         let mut module = Module::default();
-        assert_eq!(limits.validate_module(&module), Ok(()));
+        assert!(limits.validate(&module).is_ok());
 
         module.globals.push(Global {
             wasm_ty: WasmType::I32,
@@ -1223,7 +1217,7 @@ mod test {
             initializer: GlobalInit::I32Const(0),
         });
         assert_eq!(
-            limits.validate_module(&module),
+            limits.validate(&module).map_err(|e| e.to_string()),
             Err("defined globals count of 1 exceeds the limit of 0".into())
         );
     }
@@ -1247,7 +1241,7 @@ mod test {
             },
         });
         assert_eq!(
-            limits.validate_module(&module),
+            limits.validate(&module).map_err(|e| e.to_string()),
             Err(
                 "table index 0 has a minimum element size of 11 which exceeds the limit of 10"
                     .into()
@@ -1274,7 +1268,7 @@ mod test {
             offset_guard_size: 0,
         });
         assert_eq!(
-            limits.validate_module(&module),
+            limits.validate(&module).map_err(|e| e.to_string()),
             Err("memory index 0 has a minimum page size of 6 which exceeds the limit of 5".into())
         );
     }
@@ -1298,7 +1292,7 @@ mod test {
             offset_guard_size: 0,
         });
         assert_eq!(
-            limits.validate_module(&module),
+            limits.validate(&module).map_err(|e| e.to_string()),
             Err("memory index 0 has an unsupported dynamic memory plan style".into())
         );
     }
@@ -1335,7 +1329,7 @@ mod test {
 
     #[cfg(target_pointer_width = "64")]
     #[test]
-    fn test_instance_pool() -> Result<(), String> {
+    fn test_instance_pool() -> Result<()> {
         let module_limits = ModuleLimits {
             imported_functions: 0,
             imported_tables: 0,
@@ -1372,13 +1366,7 @@ mod test {
         assert_eq!(instances.instance_size, 4096);
         assert_eq!(instances.max_instances, 3);
 
-        assert_eq!(
-            &*instances
-                .free_list
-                .lock()
-                .map_err(|_| "failed to lock".to_string())?,
-            &[0, 1, 2],
-        );
+        assert_eq!(&*instances.free_list.lock().unwrap(), &[0, 1, 2],);
 
         let mut handles = Vec::new();
         let module = Arc::new(Module::default());
@@ -1409,13 +1397,7 @@ mod test {
             );
         }
 
-        assert_eq!(
-            &*instances
-                .free_list
-                .lock()
-                .map_err(|_| "failed to lock".to_string())?,
-            &[],
-        );
+        assert_eq!(&*instances.free_list.lock().unwrap(), &[],);
 
         match instances.allocate(
             PoolingAllocationStrategy::NextAvailable,
@@ -1443,20 +1425,14 @@ mod test {
             instances.deallocate(&handle);
         }
 
-        assert_eq!(
-            &*instances
-                .free_list
-                .lock()
-                .map_err(|_| "failed to lock".to_string())?,
-            &[2, 1, 0],
-        );
+        assert_eq!(&*instances.free_list.lock().unwrap(), &[2, 1, 0],);
 
         Ok(())
     }
 
     #[cfg(target_pointer_width = "64")]
     #[test]
-    fn test_memory_pool() -> Result<(), String> {
+    fn test_memory_pool() -> Result<()> {
         let pool = MemoryPool::new(
             &ModuleLimits {
                 imported_functions: 0,
@@ -1502,7 +1478,7 @@ mod test {
 
     #[cfg(target_pointer_width = "64")]
     #[test]
-    fn test_table_pool() -> Result<(), String> {
+    fn test_table_pool() -> Result<()> {
         let pool = TablePool::new(
             &ModuleLimits {
                 imported_functions: 0,
@@ -1549,7 +1525,7 @@ mod test {
 
     #[cfg(all(unix, target_pointer_width = "64"))]
     #[test]
-    fn test_stack_pool() -> Result<(), String> {
+    fn test_stack_pool() -> Result<()> {
         let pool = StackPool::new(
             &InstanceLimits {
                 count: 10,
@@ -1563,10 +1539,7 @@ mod test {
         assert_eq!(pool.page_size, 4096);
 
         assert_eq!(
-            &*pool
-                .free_list
-                .lock()
-                .map_err(|_| "failed to lock".to_string())?,
+            &*pool.free_list.lock().unwrap(),
             &[0, 1, 2, 3, 4, 5, 6, 7, 8, 9],
         );
 
@@ -1581,13 +1554,7 @@ mod test {
             stacks.push(stack);
         }
 
-        assert_eq!(
-            &*pool
-                .free_list
-                .lock()
-                .map_err(|_| "failed to lock".to_string())?,
-            &[],
-        );
+        assert_eq!(&*pool.free_list.lock().unwrap(), &[],);
 
         match pool
             .allocate(PoolingAllocationStrategy::NextAvailable)
@@ -1602,10 +1569,7 @@ mod test {
         }
 
         assert_eq!(
-            &*pool
-                .free_list
-                .lock()
-                .map_err(|_| "failed to lock".to_string())?,
+            &*pool.free_list.lock().unwrap(),
             &[9, 8, 7, 6, 5, 4, 3, 2, 1, 0],
         );
 
@@ -1624,6 +1588,7 @@ mod test {
                 },
                 4096
             )
+            .map_err(|e| e.to_string())
             .expect_err("expected a failure constructing instance allocator"),
             "the instance count limit cannot be zero"
         );
@@ -1644,6 +1609,7 @@ mod test {
                 },
                 4096
             )
+            .map_err(|e| e.to_string())
             .expect_err("expected a failure constructing instance allocator"),
             "module memory page limit of 65537 exceeds the maximum of 65536"
         );
@@ -1664,6 +1630,7 @@ mod test {
                 },
                 4096,
             )
+            .map_err(|e| e.to_string())
             .expect_err("expected a failure constructing instance allocator"),
             "module memory page limit of 2 pages exeeds the memory reservation size limit of 65536 bytes"
         );
@@ -1672,7 +1639,7 @@ mod test {
     #[cfg_attr(target_arch = "aarch64", ignore)] // https://github.com/bytecodealliance/wasmtime/pull/2518#issuecomment-747280133
     #[cfg(all(unix, target_pointer_width = "64"))]
     #[test]
-    fn test_stack_zeroed() -> Result<(), String> {
+    fn test_stack_zeroed() -> Result<()> {
         let allocator = PoolingInstanceAllocator::new(
             PoolingAllocationStrategy::NextAvailable,
             ModuleLimits {
@@ -1695,9 +1662,7 @@ mod test {
 
         unsafe {
             for _ in 0..10 {
-                let stack = allocator
-                    .allocate_fiber_stack()
-                    .map_err(|e| format!("failed to allocate stack: {}", e))?;
+                let stack = allocator.allocate_fiber_stack()?;
 
                 // The stack pointer is at the top, so decerement it first
                 let addr = stack.sub(1);
diff --git a/crates/runtime/src/instance/allocator/pooling/linux.rs b/crates/runtime/src/instance/allocator/pooling/linux.rs
index 4a8a43f08f..bd62f37dfb 100644
--- a/crates/runtime/src/instance/allocator/pooling/linux.rs
+++ b/crates/runtime/src/instance/allocator/pooling/linux.rs
@@ -1,4 +1,5 @@
 use crate::Mmap;
+use anyhow::{anyhow, Result};
 
 pub unsafe fn make_accessible(addr: *mut u8, len: usize) -> bool {
     region::protect(addr, len, region::Protection::READ_WRITE).is_ok()
@@ -16,7 +17,7 @@ pub unsafe fn decommit(addr: *mut u8, len: usize) {
     );
 }
 
-pub fn create_memory_map(accessible_size: usize, mapping_size: usize) -> Result<Mmap, String> {
+pub fn create_memory_map(accessible_size: usize, mapping_size: usize) -> Result<Mmap> {
     Mmap::accessible_reserved(accessible_size, mapping_size)
-        .map_err(|e| format!("failed to allocate pool memory: {}", e))
+        .map_err(|e| anyhow!("failed to allocate pool memory: {}", e))
 }
diff --git a/crates/runtime/src/instance/allocator/pooling/uffd.rs b/crates/runtime/src/instance/allocator/pooling/uffd.rs
index 793b63522f..b140a8d2a9 100644
--- a/crates/runtime/src/instance/allocator/pooling/uffd.rs
+++ b/crates/runtime/src/instance/allocator/pooling/uffd.rs
@@ -1,23 +1,40 @@
-//! Implements user space page fault handling with the `userfaultfd` ("uffd") system call on Linux.
+//! This module implements user space page fault handling with the `userfaultfd` ("uffd") system call on Linux.
 //!
 //! Handling page faults for memory accesses in regions relating to WebAssembly instances
-//! enables the implementation of protecting guard pages in user space rather than kernel space.
+//! enables the runtime to protect guard pages in user space rather than kernel space (i.e. without `mprotect`).
 //!
-//! This reduces the number of system calls and kernel locks needed to provide correct
-//! WebAssembly memory semantics.
+//! Additionally, linear memories can be lazy-initialized upon first access.
 //!
-//! Additionally, linear memories can be lazy-initialized upon access.
+//! Handling faults in user space is slower than handling faults in the kernel. However,
+//! in use cases where there is a high number of concurrently executing instances, handling the faults
+//! in user space requires rarely changing memory protection levels.  This can improve concurrency
+//! by not taking kernel memory manager locks and may decrease TLB shootdowns as fewer page table entries need
+//! to continually change.
+//!
+//! Here's how the `uffd` feature works:
+//!
+//! 1. A user fault file descriptor is created to monitor specific areas of the address space.
+//! 2. A thread is spawned to continually read events from the user fault file descriptor.
+//! 3. When a page fault event is received, the handler thread calculates where the fault occurred:
+//!    a) If the fault occurs on a table page, it is handled by zeroing the page.
+//!    b) If the fault occurs on a linear memory page, it is handled by either copying the page from
+//!       initialization data or zeroing it.
+//!    c) If the fault occurs on a guard page, the protection level of the guard page is changed to
+//!       force the kernel to signal SIGSEV on the next retry. The faulting page is recorded so the
+//!       protection level can be reset in the future.
+//! 4. Faults to address space relating to an instance may occur from both Wasmtime (e.g. instance
+//!    initialization) or from WebAssembly code (e.g. reading from or writing to linear memory),
+//!    therefore the user fault handling must do as little work as possible to handle the fault.
+//! 5. When the pooling allocator is dropped, it will drop the memory mappings relating to the pool; this
+//!    generates unmap events for the fault handling thread, which responds by decrementing the mapping
+//!    count. When the count reaches zero, the user fault handling thread will gracefully terminate.
 //!
 //! This feature requires a Linux kernel 4.11 or newer to use.
 
-use super::{InstancePool, StackPool};
+use super::InstancePool;
 use crate::{instance::Instance, Mmap};
-use std::convert::TryInto;
+use anyhow::{bail, Context, Result};
 use std::ptr;
-use std::sync::{
-    atomic::{AtomicBool, Ordering},
-    Arc,
-};
 use std::thread;
 use userfaultfd::{Event, FeatureFlags, IoctlFlags, Uffd, UffdBuilder};
 use wasmtime_environ::{entity::EntityRef, wasm::DefinedMemoryIndex, MemoryInitialization};
@@ -45,11 +62,11 @@ pub unsafe fn decommit(addr: *mut u8, len: usize) {
     );
 }
 
-pub fn create_memory_map(_accessible_size: usize, mapping_size: usize) -> Result<Mmap, String> {
+pub fn create_memory_map(_accessible_size: usize, mapping_size: usize) -> Result<Mmap> {
     // Allocate a single read-write region at once
     // As writable pages need to count towards commit charge, use MAP_NORESERVE to override.
-    // This implies that the kernel is configured to allow overcommit or else
-    // this allocation will almost certainly fail without a plethora of physical memory to back the allocation.
+    // This implies that the kernel is configured to allow overcommit or else this allocation
+    // will almost certainly fail without a plethora of physical memory to back the allocation.
     // The consequence of not reserving is that our process may segfault on any write to a memory
     // page that cannot be backed (i.e. out of memory conditions).
 
@@ -68,10 +85,10 @@ pub fn create_memory_map(_accessible_size: usize, mapping_size: usize) -> Result
         );
 
         if ptr as isize == -1_isize {
-            return Err(format!(
-                "failed to allocate pool memory: {}",
+            bail!(
+                "failed to allocate pool memory: mmap failed with {}",
                 std::io::Error::last_os_error()
-            ));
+            );
         }
 
         Ok(Mmap::from_raw(ptr as usize, mapping_size))
@@ -98,22 +115,10 @@ enum AddressLocation<'a> {
         /// The instance related to the memory page that was accessed.
         instance: &'a Instance,
         /// The index of the memory that was accessed.
-        memory_index: usize,
+        memory_index: DefinedMemoryIndex,
         /// The Wasm page index to initialize if the access was not a guard page.
         page_index: Option<usize>,
     },
-    /// The address location is in an execution stack.
-    /// The fault handler will zero the page.
-    StackPage {
-        /// The address of the page being accessed.
-        page_addr: *mut u8,
-        /// The length of the page being accessed.
-        len: usize,
-        /// The index of the stack that was accessed.
-        index: usize,
-        /// Whether or not the access was to a guard page.
-        guard_page: bool,
-    },
 }
 
 /// Used to resolve fault addresses to address locations.
@@ -132,22 +137,16 @@ struct AddressLocator {
     tables_start: usize,
     tables_end: usize,
     table_size: usize,
-    stacks_start: usize,
-    stacks_end: usize,
-    stack_size: usize,
     page_size: usize,
 }
 
 impl AddressLocator {
-    fn new(instances: &InstancePool, stacks: &StackPool) -> Self {
+    fn new(instances: &InstancePool) -> Self {
         let instances_start = instances.mapping.as_ptr() as usize;
         let memories_start = instances.memories.mapping.as_ptr() as usize;
         let memories_end = memories_start + instances.memories.mapping.len();
         let tables_start = instances.tables.mapping.as_ptr() as usize;
         let tables_end = tables_start + instances.tables.mapping.len();
-        let stacks_start = stacks.mapping.as_ptr() as usize;
-        let stacks_end = stacks_start + stacks.mapping.len();
-        let stack_size = stacks.stack_size;
 
         // Should always have instances
         debug_assert!(instances_start != 0);
@@ -163,9 +162,6 @@ impl AddressLocator {
             tables_start,
             tables_end,
             table_size: instances.tables.table_size,
-            stacks_start,
-            stacks_end,
-            stack_size,
             page_size: instances.tables.page_size,
         }
     }
@@ -191,25 +187,18 @@ impl AddressLocator {
         // Check for a memory location
         if addr >= self.memories_start && addr < self.memories_end {
             let index = (addr - self.memories_start) / self.memory_size;
-            let memory_index = index % self.max_memories;
+            let memory_index = DefinedMemoryIndex::new(index % self.max_memories);
             let memory_start = self.memories_start + (index * self.memory_size);
             let page_index = (addr - memory_start) / WASM_PAGE_SIZE;
             let instance = self.get_instance(index / self.max_memories);
 
-            let init_page_index = instance
-                .memories
-                .get(
-                    DefinedMemoryIndex::from_u32(memory_index as u32)
-                        .try_into()
-                        .unwrap(),
-                )
-                .and_then(|m| {
-                    if page_index < m.size() as usize {
-                        Some(page_index)
-                    } else {
-                        None
-                    }
-                });
+            let init_page_index = instance.memories.get(memory_index).and_then(|m| {
+                if page_index < m.size() as usize {
+                    Some(page_index)
+                } else {
+                    None
+                }
+            });
 
             return Some(AddressLocation::MemoryPage {
                 page_addr: (memory_start + page_index * WASM_PAGE_SIZE) as _,
@@ -233,128 +222,125 @@ impl AddressLocator {
             });
         }
 
-        // Check for a stack location
-        if addr >= self.stacks_start && addr < self.stacks_end {
-            let index = (addr - self.stacks_start) / self.stack_size;
-            let stack_start = self.stacks_start + (index * self.stack_size);
-            let stack_offset = addr - stack_start;
-            let page_offset = (stack_offset / self.page_size) * self.page_size;
-
-            return Some(AddressLocation::StackPage {
-                page_addr: (stack_start + page_offset) as _,
-                len: self.page_size,
-                index,
-                guard_page: stack_offset < self.page_size,
-            });
-        }
-
         None
     }
 }
 
-unsafe fn wake_guard_page_access(
-    uffd: &Uffd,
-    page_addr: *const u8,
-    len: usize,
-) -> Result<(), String> {
-    // Set the page to NONE to induce a SIGSEV for the access on the next retry
+/// This is called following a fault on a guard page.
+///
+/// Because the region being monitored is protected read-write, this needs to set the
+/// protection level to `NONE` before waking the page.
+///
+/// This will cause the kernel to raise a SIGSEGV when retrying the fault.
+unsafe fn wake_guard_page_access(uffd: &Uffd, page_addr: *const u8, len: usize) -> Result<()> {
+    // Set the page to NONE to induce a SIGSEGV for the access on the next retry
     region::protect(page_addr, len, region::Protection::NONE)
-        .map_err(|e| format!("failed to change guard page protection: {}", e))?;
+        .context("failed to change guard page protection")?;
 
-    uffd.wake(page_addr as _, len).map_err(|e| {
-        format!(
-            "failed to wake page at {:p} with length {}: {}",
-            page_addr, len, e
-        )
-    })?;
+    uffd.wake(page_addr as _, len)
+        .context("failed to wake guard page access")?;
 
     Ok(())
 }
 
+/// This is called to initialize a linear memory page (64 KiB).
+///
+/// If paged initialization is used for the module, then we can instruct the kernel to back the page with
+/// what is already stored in the initialization data; if the page isn't in the initialization data,
+/// it will be zeroed instead.
+///
+/// If paged initialization isn't being used, we zero the page. Initialization happens
+/// at module instantiation in this case and the segment data will be then copied to the zeroed page.
 unsafe fn initialize_wasm_page(
     uffd: &Uffd,
     instance: &Instance,
     page_addr: *const u8,
-    memory_index: usize,
+    memory_index: DefinedMemoryIndex,
     page_index: usize,
-) -> Result<(), String> {
-    if let Some(MemoryInitialization::Paged { page_size, map }) =
-        &instance.module.memory_initialization
-    {
-        let memory_index = DefinedMemoryIndex::new(memory_index);
-        let memory = instance.memory(memory_index);
+) -> Result<()> {
+    // Check for paged initialization and copy the page if present in the initialization data
+    if let MemoryInitialization::Paged { map, .. } = &instance.module.memory_initialization {
         let pages = &map[memory_index];
-        debug_assert_eq!(WASM_PAGE_SIZE % page_size, 0);
 
-        let count = WASM_PAGE_SIZE / page_size;
-        let start = page_index * count;
+        if let Some(Some(data)) = pages.get(page_index) {
+            debug_assert_eq!(data.len(), WASM_PAGE_SIZE);
 
-        for i in start..start + count {
-            let dst = memory.base.add(i * page_size);
+            log::trace!(
+                "copying linear memory page from {:p} to {:p}",
+                data.as_ptr(),
+                page_addr
+            );
 
-            match pages.get(i) {
-                Some(Some(data)) => {
-                    log::trace!(
-                        "copying page initialization data from {:p} to {:p} with length {}",
-                        data,
-                        dst,
-                        page_size
-                    );
+            uffd.copy(data.as_ptr() as _, page_addr as _, WASM_PAGE_SIZE, true)
+                .context("failed to copy linear memory page")?;
 
-                    // Copy the page data without waking
-                    uffd.copy(data.as_ptr() as _, dst as _, *page_size, false)
-                        .map_err(|e| {
-                            format!(
-                                "failed to copy page from {:p} to {:p} with length {}: {}",
-                                data, dst, page_size, e
-                            )
-                        })?;
+            return Ok(());
+        }
+    }
+
+    log::trace!("zeroing linear memory page at {:p}", page_addr);
+
+    uffd.zeropage(page_addr as _, WASM_PAGE_SIZE, true)
+        .context("failed to zero linear memory page")?;
+
+    Ok(())
+}
+
+unsafe fn handle_page_fault(
+    uffd: &Uffd,
+    locator: &AddressLocator,
+    addr: *mut std::ffi::c_void,
+) -> Result<()> {
+    match locator.get_location(addr as usize) {
+        Some(AddressLocation::TablePage { page_addr, len }) => {
+            log::trace!(
+                "handling fault in table at address {:p} on page {:p}",
+                addr,
+                page_addr,
+            );
+
+            // Tables are always initialized upon instantiation, so zero the page
+            uffd.zeropage(page_addr as _, len, true)
+                .context("failed to zero table page")?;
+        }
+        Some(AddressLocation::MemoryPage {
+            page_addr,
+            len,
+            instance,
+            memory_index,
+            page_index,
+        }) => {
+            log::trace!(
+                "handling fault in linear memory at address {:p} on page {:p}",
+                addr,
+                page_addr
+            );
+
+            match page_index {
+                Some(page_index) => {
+                    initialize_wasm_page(&uffd, instance, page_addr, memory_index, page_index)?;
                 }
-                _ => {
-                    log::trace!("zeroing page at {:p} with length {}", dst, page_size);
+                None => {
+                    log::trace!("out of bounds memory access at {:p}", addr);
 
-                    // No data, zero the page without waking
-                    uffd.zeropage(dst as _, *page_size, false).map_err(|e| {
-                        format!(
-                            "failed to zero page at {:p} with length {}: {}",
-                            dst, page_size, e
-                        )
-                    })?;
+                    // Record the guard page fault with the instance so it can be reset later.
+                    instance.record_guard_page_fault(page_addr, len, reset_guard_page);
+                    wake_guard_page_access(&uffd, page_addr, len)?;
                 }
             }
         }
-
-        // Finally wake the entire wasm page
-        uffd.wake(page_addr as _, WASM_PAGE_SIZE).map_err(|e| {
-            format!(
-                "failed to wake page at {:p} with length {}: {}",
-                page_addr, WASM_PAGE_SIZE, e
-            )
-        })
-    } else {
-        log::trace!(
-            "initialization data is not paged; zeroing Wasm page at {:p}",
-            page_addr
-        );
-
-        uffd.zeropage(page_addr as _, WASM_PAGE_SIZE, true)
-            .map_err(|e| {
-                format!(
-                    "failed to zero page at {:p} with length {}: {}",
-                    page_addr, WASM_PAGE_SIZE, e
-                )
-            })?;
-
-        Ok(())
+        None => {
+            bail!(
+                "failed to locate fault address {:p} in registered memory regions",
+                addr
+            );
+        }
     }
+
+    Ok(())
 }
 
-fn handler_thread(
-    uffd: Uffd,
-    locator: AddressLocator,
-    mut registrations: usize,
-    faulted_stack_guard_pages: Arc<[AtomicBool]>,
-) -> Result<(), String> {
+fn handler_thread(uffd: Uffd, locator: AddressLocator, mut registrations: usize) -> Result<()> {
     loop {
         match uffd.read_event().expect("failed to read event") {
             Some(Event::Unmap { start, end }) => {
@@ -364,7 +350,6 @@ fn handler_thread(
 
                 if (start == locator.memories_start && end == locator.memories_end)
                     || (start == locator.tables_start && end == locator.tables_end)
-                    || (start == locator.stacks_start && end == locator.stacks_end)
                 {
                     registrations -= 1;
                     if registrations == 0 {
@@ -374,104 +359,11 @@ fn handler_thread(
                     panic!("unexpected memory region unmapped");
                 }
             }
-            Some(Event::Pagefault {
-                addr: access_addr, ..
-            }) => {
-                unsafe {
-                    match locator.get_location(access_addr as usize) {
-                        Some(AddressLocation::TablePage { page_addr, len }) => {
-                            log::trace!(
-                                "handling fault in table at address {:p} on page {:p}",
-                                access_addr,
-                                page_addr,
-                            );
-
-                            // Tables are always initialized upon instantiation, so zero the page
-                            uffd.zeropage(page_addr as _, len, true).map_err(|e| {
-                                format!(
-                                    "failed to zero page at {:p} with length {}: {}",
-                                    page_addr, len, e
-                                )
-                            })?;
-                        }
-                        Some(AddressLocation::MemoryPage {
-                            page_addr,
-                            len,
-                            instance,
-                            memory_index,
-                            page_index,
-                        }) => {
-                            log::trace!(
-                                "handling fault in linear memory at address {:p} on page {:p}",
-                                access_addr,
-                                page_addr
-                            );
-
-                            match page_index {
-                                Some(page_index) => {
-                                    initialize_wasm_page(
-                                        &uffd,
-                                        instance,
-                                        page_addr,
-                                        memory_index,
-                                        page_index,
-                                    )?;
-                                }
-                                None => {
-                                    log::trace!("out of bounds memory access at {:p}", access_addr);
-
-                                    // Record the guard page fault with the instance so it can be reset later.
-                                    instance.record_guard_page_fault(
-                                        page_addr,
-                                        len,
-                                        reset_guard_page,
-                                    );
-                                    wake_guard_page_access(&uffd, page_addr, len)?;
-                                }
-                            }
-                        }
-                        Some(AddressLocation::StackPage {
-                            page_addr,
-                            len,
-                            index,
-                            guard_page,
-                        }) => {
-                            log::trace!(
-                                "handling fault in stack {} at address {:p}",
-                                index,
-                                access_addr,
-                            );
-
-                            if guard_page {
-                                // Logging as trace as stack guard pages might be a trap condition in the future
-                                log::trace!("stack overflow fault at {:p}", access_addr);
-
-                                // Mark the stack as having a faulted guard page
-                                // The next time the stack is used the guard page will be reset
-                                faulted_stack_guard_pages[index].store(true, Ordering::SeqCst);
-                                wake_guard_page_access(&uffd, page_addr, len)?;
-                                continue;
-                            }
-
-                            // Always zero stack pages
-                            uffd.zeropage(page_addr as _, len, true).map_err(|e| {
-                                format!(
-                                    "failed to zero page at {:p} with length {}: {}",
-                                    page_addr, len, e
-                                )
-                            })?;
-                        }
-                        None => {
-                            return Err(format!(
-                                "failed to locate fault address {:p} in registered memory regions",
-                                access_addr
-                            ));
-                        }
-                    }
-                }
-            }
+            Some(Event::Pagefault { addr, .. }) => unsafe {
+                handle_page_fault(&uffd, &locator, addr as _)?
+            },
             Some(_) => continue,
-            None => break,
+            None => bail!("no event was read from the user fault descriptor"),
         }
     }
 
@@ -482,16 +374,16 @@ fn handler_thread(
 
 #[derive(Debug)]
 pub struct PageFaultHandler {
-    thread: Option<thread::JoinHandle<Result<(), String>>>,
+    thread: Option<thread::JoinHandle<Result<()>>>,
 }
 
 impl PageFaultHandler {
-    pub(super) fn new(instances: &InstancePool, stacks: &StackPool) -> Result<Self, String> {
+    pub(super) fn new(instances: &InstancePool) -> Result<Self> {
         let uffd = UffdBuilder::new()
             .close_on_exec(true)
             .require_features(FeatureFlags::EVENT_UNMAP)
             .create()
-            .map_err(|e| format!("failed to create user fault descriptor: {}", e))?;
+            .context("failed to create user fault descriptor")?;
 
         // Register the ranges with the userfault fd
         let mut registrations = 0;
@@ -504,7 +396,6 @@ impl PageFaultHandler {
                 instances.tables.mapping.as_ptr() as usize,
                 instances.tables.mapping.len(),
             ),
-            (stacks.mapping.as_ptr() as usize, stacks.mapping.len()),
         ] {
             if *start == 0 || *len == 0 {
                 continue;
@@ -512,13 +403,13 @@ impl PageFaultHandler {
 
             let ioctls = uffd
                 .register(*start as _, *len)
-                .map_err(|e| format!("failed to register user fault range: {}", e))?;
+                .context("failed to register user fault range")?;
 
             if !ioctls.contains(IoctlFlags::WAKE | IoctlFlags::COPY | IoctlFlags::ZEROPAGE) {
-                return Err(format!(
+                bail!(
                     "required user fault ioctls not supported; found: {:?}",
                     ioctls,
-                ));
+                );
             }
 
             registrations += 1;
@@ -533,17 +424,13 @@ impl PageFaultHandler {
                 registrations
             );
 
-            let locator = AddressLocator::new(&instances, &stacks);
-
-            let faulted_stack_guard_pages = stacks.faulted_guard_pages.clone();
+            let locator = AddressLocator::new(&instances);
 
             Some(
                 thread::Builder::new()
                     .name("page fault handler".into())
-                    .spawn(move || {
-                        handler_thread(uffd, locator, registrations, faulted_stack_guard_pages)
-                    })
-                    .map_err(|e| format!("failed to spawn page fault handler thread: {}", e))?,
+                    .spawn(move || handler_thread(uffd, locator, registrations))
+                    .context("failed to spawn page fault handler thread")?,
             )
         };
 
@@ -553,6 +440,9 @@ impl PageFaultHandler {
 
 impl Drop for PageFaultHandler {
     fn drop(&mut self) {
+        // The handler thread should terminate once all monitored regions of memory are unmapped.
+        // The pooling instance allocator ensures that the regions are unmapped prior to dropping
+        // the user fault handler.
         if let Some(thread) = self.thread.take() {
             thread
                 .join()
@@ -569,6 +459,7 @@ mod test {
         table::max_table_element_size, Imports, InstanceAllocationRequest, InstanceLimits,
         ModuleLimits, PoolingAllocationStrategy, VMSharedSignatureIndex,
     };
+    use std::sync::Arc;
     use wasmtime_environ::{
         entity::PrimaryMap,
         wasm::{Memory, Table, TableElementType, WasmType},
@@ -598,9 +489,8 @@ mod test {
 
         let instances =
             InstancePool::new(&module_limits, &instance_limits).expect("should allocate");
-        let stacks = StackPool::new(&instance_limits, 8192).expect("should allocate");
 
-        let locator = AddressLocator::new(&instances, &stacks);
+        let locator = AddressLocator::new(&instances);
 
         assert_eq!(locator.instances_start, instances.mapping.as_ptr() as usize);
         assert_eq!(locator.instance_size, 4096);
@@ -625,20 +515,10 @@ mod test {
         );
         assert_eq!(locator.table_size, 8192);
 
-        assert_eq!(locator.stacks_start, stacks.mapping.as_ptr() as usize);
-        assert_eq!(
-            locator.stacks_end,
-            locator.stacks_start + stacks.mapping.len()
-        );
-        assert_eq!(locator.stack_size, 12288);
-
         unsafe {
             assert!(locator.get_location(0).is_none());
             assert!(locator
-                .get_location(std::cmp::max(
-                    locator.memories_end,
-                    std::cmp::max(locator.tables_end, locator.stacks_end)
-                ))
+                .get_location(std::cmp::max(locator.memories_end, locator.tables_end))
                 .is_none());
 
             let mut module = Module::new();
@@ -667,9 +547,7 @@ mod test {
                 });
             }
 
-            module_limits
-                .validate_module(&module)
-                .expect("should validate");
+            module_limits.validate(&module).expect("should validate");
 
             let mut handles = Vec::new();
             let module = Arc::new(module);
@@ -719,7 +597,7 @@ mod test {
                         }) => {
                             assert_eq!(page_addr, memory_start as _);
                             assert_eq!(len, WASM_PAGE_SIZE);
-                            assert_eq!(mem_index, memory_index);
+                            assert_eq!(mem_index, DefinedMemoryIndex::new(memory_index));
                             assert_eq!(page_index, Some(0));
                         }
                         _ => panic!("expected a memory page location"),
@@ -736,7 +614,7 @@ mod test {
                         }) => {
                             assert_eq!(page_addr, (memory_start + WASM_PAGE_SIZE) as _);
                             assert_eq!(len, WASM_PAGE_SIZE);
-                            assert_eq!(mem_index, memory_index);
+                            assert_eq!(mem_index, DefinedMemoryIndex::new(memory_index));
                             assert_eq!(page_index, Some(1));
                         }
                         _ => panic!("expected a memory page location"),
@@ -753,7 +631,7 @@ mod test {
                         }) => {
                             assert_eq!(page_addr, (memory_start + (9 * WASM_PAGE_SIZE)) as _);
                             assert_eq!(len, WASM_PAGE_SIZE);
-                            assert_eq!(mem_index, memory_index);
+                            assert_eq!(mem_index, DefinedMemoryIndex::new(memory_index));
                             assert_eq!(page_index, None);
                         }
                         _ => panic!("expected a memory page location"),
@@ -788,43 +666,6 @@ mod test {
                 }
             }
 
-            // Validate stack locations
-            for stack_index in 0..instances.max_instances {
-                let stack_start = locator.stacks_start + (stack_index * locator.stack_size);
-
-                // Check for stack page location
-                match locator.get_location(stack_start + locator.page_size * 2) {
-                    Some(AddressLocation::StackPage {
-                        page_addr,
-                        len,
-                        index,
-                        guard_page,
-                    }) => {
-                        assert_eq!(page_addr, (stack_start + locator.page_size * 2) as _);
-                        assert_eq!(len, locator.page_size);
-                        assert_eq!(index, stack_index);
-                        assert!(!guard_page);
-                    }
-                    _ => panic!("expected a stack page location"),
-                }
-
-                // Check for guard page
-                match locator.get_location(stack_start) {
-                    Some(AddressLocation::StackPage {
-                        page_addr,
-                        len,
-                        index,
-                        guard_page,
-                    }) => {
-                        assert_eq!(page_addr, stack_start as _);
-                        assert_eq!(len, locator.page_size);
-                        assert_eq!(index, stack_index);
-                        assert!(guard_page);
-                    }
-                    _ => panic!("expected a stack page location"),
-                }
-            }
-
             for handle in handles.drain(..) {
                 instances.deallocate(&handle);
             }
diff --git a/crates/runtime/src/instance/allocator/pooling/unix.rs b/crates/runtime/src/instance/allocator/pooling/unix.rs
index 900e73d174..9cc68b3361 100644
--- a/crates/runtime/src/instance/allocator/pooling/unix.rs
+++ b/crates/runtime/src/instance/allocator/pooling/unix.rs
@@ -1,4 +1,5 @@
 use crate::Mmap;
+use anyhow::{anyhow, Result};
 
 pub unsafe fn make_accessible(addr: *mut u8, len: usize) -> bool {
     region::protect(addr, len, region::Protection::READ_WRITE).is_ok()
@@ -20,7 +21,7 @@ pub unsafe fn decommit(addr: *mut u8, len: usize) {
     );
 }
 
-pub fn create_memory_map(accessible_size: usize, mapping_size: usize) -> Result<Mmap, String> {
+pub fn create_memory_map(accessible_size: usize, mapping_size: usize) -> Result<Mmap> {
     Mmap::accessible_reserved(accessible_size, mapping_size)
-        .map_err(|e| format!("failed to allocate pool memory: {}", e))
+        .map_err(|e| anyhow!("failed to allocate pool memory: {}", e))
 }
diff --git a/crates/runtime/src/instance/allocator/pooling/windows.rs b/crates/runtime/src/instance/allocator/pooling/windows.rs
index fe8566558a..159f00b63f 100644
--- a/crates/runtime/src/instance/allocator/pooling/windows.rs
+++ b/crates/runtime/src/instance/allocator/pooling/windows.rs
@@ -1,4 +1,5 @@
 use crate::Mmap;
+use anyhow::{anyhow, Result};
 use winapi::um::memoryapi::{VirtualAlloc, VirtualFree};
 use winapi::um::winnt::{MEM_COMMIT, MEM_DECOMMIT, PAGE_READWRITE};
 
@@ -15,7 +16,7 @@ pub unsafe fn decommit(addr: *mut u8, len: usize) {
     );
 }
 
-pub fn create_memory_map(accessible_size: usize, mapping_size: usize) -> Result<Mmap, String> {
+pub fn create_memory_map(accessible_size: usize, mapping_size: usize) -> Result<Mmap> {
     Mmap::accessible_reserved(accessible_size, mapping_size)
-        .map_err(|e| format!("failed to allocate pool memory: {}", e))
+        .map_err(|e| anyhow!("failed to allocate pool memory: {}", e))
 }
diff --git a/crates/wasmtime/src/config.rs b/crates/wasmtime/src/config.rs
index 77d5c52be7..30e63546e8 100644
--- a/crates/wasmtime/src/config.rs
+++ b/crates/wasmtime/src/config.rs
@@ -627,15 +627,12 @@ impl Config {
                 #[cfg(not(feature = "async"))]
                 let stack_size = 0;
 
-                Some(Arc::new(
-                    PoolingInstanceAllocator::new(
-                        strategy,
-                        module_limits,
-                        instance_limits,
-                        stack_size,
-                    )
-                    .map_err(|e| anyhow::anyhow!(e))?,
-                ))
+                Some(Arc::new(PoolingInstanceAllocator::new(
+                    strategy,
+                    module_limits,
+                    instance_limits,
+                    stack_size,
+                )?))
             }
         };
         Ok(self)
diff --git a/crates/wasmtime/src/module.rs b/crates/wasmtime/src/module.rs
index a9d32ee1d1..f87b28f6fe 100644
--- a/crates/wasmtime/src/module.rs
+++ b/crates/wasmtime/src/module.rs
@@ -307,22 +307,36 @@ impl Module {
     /// # }
     /// ```
     pub fn from_binary(engine: &Engine, binary: &[u8]) -> Result<Module> {
-        // Check with the instance allocator to see if the given module is supported
-        let allocator = engine.config().instance_allocator();
+        cfg_if::cfg_if! {
+            if #[cfg(feature = "cache")] {
+                let (main_module, artifacts, types) = ModuleCacheEntry::new(
+                    "wasmtime",
+                    engine.cache_config(),
+                )
+                .get_data((engine.compiler(), binary), |(compiler, binary)| {
+                    cfg_if::cfg_if! {
+                        if #[cfg(all(feature = "uffd", target_os = "linux"))] {
+                            let use_paged_mem_init = true;
+                        } else {
+                            let use_paged_mem_init = false;
+                        }
+                    };
 
-        #[cfg(feature = "cache")]
-        let (main_module, artifacts, types) = ModuleCacheEntry::new(
-            "wasmtime",
-            engine.cache_config(),
-        )
-        .get_data((engine.compiler(), binary), |(compiler, binary)| {
-            CompilationArtifacts::build(compiler, binary, |m| allocator.validate_module(m))
-        })?;
-        #[cfg(not(feature = "cache"))]
-        let (main_module, artifacts, types) =
-            CompilationArtifacts::build(engine.compiler(), binary, |m| {
-                allocator.validate_module(m)
-            })?;
+                    CompilationArtifacts::build(compiler, binary, use_paged_mem_init)
+                })?;
+            } else {
+                cfg_if::cfg_if! {
+                    if #[cfg(all(feature = "uffd", target_os = "linux"))] {
+                        let use_paged_mem_init = true;
+                    } else {
+                        let use_paged_mem_init = false;
+                    }
+                };
+
+                let (main_module, artifacts, types) =
+                    CompilationArtifacts::build(engine.compiler(), binary, use_paged_mem_init)?;
+            }
+        };
 
         let mut modules = CompiledModule::from_artifacts_list(
             artifacts,
@@ -331,6 +345,12 @@ impl Module {
         )?;
         let module = modules.remove(main_module);
 
+        // Validate the module can be used with the current allocator
+        engine
+            .config()
+            .instance_allocator()
+            .validate(module.module())?;
+
         Ok(Module {
             inner: Arc::new(ModuleInner {
                 engine: engine.clone(),
diff --git a/tests/all/pooling_allocator.rs b/tests/all/pooling_allocator.rs
index d7b7101fee..bdad287aa9 100644
--- a/tests/all/pooling_allocator.rs
+++ b/tests/all/pooling_allocator.rs
@@ -48,7 +48,10 @@ fn memory_limit() -> Result<()> {
     // Module should fail to validate because the minimum is greater than the configured limit
     match Module::new(&engine, r#"(module (memory 4))"#) {
         Ok(_) => panic!("module compilation should fail"),
-        Err(e) => assert_eq!(e.to_string(), "Validation error: memory index 0 has a minimum page size of 4 which exceeds the limit of 3")
+        Err(e) => assert_eq!(
+            e.to_string(),
+            "memory index 0 has a minimum page size of 4 which exceeds the limit of 3"
+        ),
     }
 
     let module = Module::new(
@@ -243,7 +246,10 @@ fn table_limit() -> Result<()> {
     // Module should fail to validate because the minimum is greater than the configured limit
     match Module::new(&engine, r#"(module (table 31 funcref))"#) {
         Ok(_) => panic!("module compilation should fail"),
-        Err(e) => assert_eq!(e.to_string(), "Validation error: table index 0 has a minimum element size of 31 which exceeds the limit of 10")
+        Err(e) => assert_eq!(
+            e.to_string(),
+            "table index 0 has a minimum element size of 31 which exceeds the limit of 10"
+        ),
     }
 
     let module = Module::new(