Refactor how concurrency support is enabled in a Store (#12416)

* Document panics from using CM async machinery when CM async is not enabled

* Refactor how concurrency support is enabled in a `Store`

This commit is an extension/refactor of #12377 and #12379. Notably this
decouples the runtime behavior of Wasmtime from enabled/disabled
WebAssembly proposals. This enables the `wasmtime serve` subcommand, for
example, to continue to disallow component-model-async by default but
continue to use `*_concurrent` under the hood.

Specifically a new `Config::concurrency_support` knob is added. This is
plumbed directly through to `Tunables` and takes over the preexisting
`component_model_concurrency` field. This field configures whether
tasks/etc are enabled at runtime for component-y things. The default
value of this configuration option is the same as `cfg!(feature =
"component-model-async")`, and this field is required if
component-model-async wasm proposals are enabled. It's intended that
eventually this'll affect on-by-default wasm features in Wasmtime
depending if the support is compiled in.

This results in a subtle shift in behavior where component-model-async
concurrency is used by default now because the feature is turned on by
default, even though the wasm features are off-by-default. This required
adjusting a few indices expected in runtime tests due to tasks/threads
being allocated in index spaces.

Finally, this additionally denies access at runtime to
`Linker::*_concurrent` when concurrent support is disabled as otherwise
the various runtime data structures won't be initialized and panics will
happen.

Closes #12393

* Add a `-Wconcurrency-support` CLI flag

Used to update disas tests to show that, when disabled, old codegen
quality is preserved

* Ungate `Config` flag

* Review comments

---------

Co-authored-by: Nick Fitzgerald <fitzgen@gmail.com>

Author: alexcrichton

crates/cli-flags/src/lib.rs

@@ -412,6 +412,9 @@ wasmtime_option_group! {
         /// Component model support for fixed-length lists: this corresponds
         /// to the 🔧 emoji in the component model specification
         pub component_model_fixed_length_lists: Option<bool>,
+        /// Whether or not any concurrency infrastructure in Wasmtime is
+        /// enabled or not.
+        pub concurrency_support: Option<bool>,
     }
 
     enum Wasm {
@@ -1006,6 +1009,10 @@ impl CommonOptions {
             config.gc_support(enable);
         }
 
+        if let Some(enable) = self.wasm.concurrency_support {
+            config.concurrency_support(enable);
+        }
+
         if let Some(enable) = self.wasm.shared_memory {
             config.shared_memory(enable);
         }

crates/cranelift/src/compiler/component.rs

@@ -1220,7 +1220,7 @@ impl<'a> TrampolineCompiler<'a> {
                     .ins()
                     .trapz(masked, TRAP_CANNOT_LEAVE_COMPONENT);
 
-                if self.compiler.tunables.component_model_concurrency {
+                if self.compiler.tunables.concurrency_support {
                     // Stash the old value of `may_block` and then set it to false.
                     let old_may_block = self.builder.ins().load(
                         ir::types::I32,

crates/environ/src/fact/trampoline.rs

@@ -180,7 +180,7 @@ pub(super) fn compile(module: &mut Module<'_>, adapter: &AdapterData) {
             compiler.compile_sync_to_sync_adapter(adapter, &lower_sig, &lift_sig)
         }
         (true, true) => {
-            assert!(module.tunables.component_model_concurrency);
+            assert!(module.tunables.concurrency_support);
 
             // In the async->async case, we must compile a couple of helper functions:
             //
@@ -209,7 +209,7 @@ pub(super) fn compile(module: &mut Module<'_>, adapter: &AdapterData) {
             );
         }
         (false, true) => {
-            assert!(module.tunables.component_model_concurrency);
+            assert!(module.tunables.concurrency_support);
 
             // Like the async->async case above, for the sync->async case we
             // also need `async-start` and `async-return` helper functions to
@@ -235,7 +235,7 @@ pub(super) fn compile(module: &mut Module<'_>, adapter: &AdapterData) {
             );
         }
         (true, false) => {
-            assert!(module.tunables.component_model_concurrency);
+            assert!(module.tunables.concurrency_support);
 
             // As with the async->async and sync->async cases above, for the
             // async->sync case we use `async-start` and `async-return` helper
@@ -759,7 +759,7 @@ impl<'a, 'b> Compiler<'a, 'b> {
             Trap::CannotLeaveComponent,
         );
 
-        let old_task_may_block = if self.module.tunables.component_model_concurrency {
+        let old_task_may_block = if self.module.tunables.concurrency_support {
             // Save, clear, and later restore the `may_block` field.
             let task_may_block = self.module.import_task_may_block();
             let old_task_may_block = if self.types[adapter.lift.ty].async_ {
@@ -871,7 +871,7 @@ impl<'a, 'b> Compiler<'a, 'b> {
             self.instruction(Call(exit.as_u32()));
         }
 
-        if self.module.tunables.component_model_concurrency {
+        if self.module.tunables.concurrency_support {
             // Pop the task we pushed earlier off of the current task stack.
             //
             // FIXME: Apply the optimizations described in #12311.

crates/environ/src/tunables.rs

@@ -143,7 +143,7 @@ define_tunables! {
 
         /// Whether any component model feature related to concurrency is
         /// enabled.
-        pub component_model_concurrency: bool,
+        pub concurrency_support: bool,
     }
 
     pub struct ConfigTunables {
@@ -219,7 +219,7 @@ impl Tunables {
             inlining_small_callee_size: 50,
             inlining_sum_size_threshold: 2000,
             debug_guest: false,
-            component_model_concurrency: true,
+            concurrency_support: true,
         }
     }
 

crates/wasmtime/src/config.rs

@@ -2358,6 +2358,10 @@ impl Config {
 
         let mut tunables = Tunables::default_for_target(&self.compiler_target())?;
 
+        // By default this is enabled with the Cargo feature, and if the feature
+        // is missing this is disabled.
+        tunables.concurrency_support = cfg!(feature = "component-model-async");
+
         // If no target is explicitly specified then further refine `tunables`
         // for the configuration of this host depending on what platform
         // features were found available at compile time. This means that anyone
@@ -2430,9 +2434,23 @@ impl Config {
             );
         }
 
-        #[cfg(feature = "component-model")]
-        {
-            tunables.component_model_concurrency = self.cm_concurrency_enabled();
+        // Concurrency support is required for some component model features.
+        let requires_concurrency = WasmFeatures::CM_ASYNC
+            | WasmFeatures::CM_ASYNC_BUILTINS
+            | WasmFeatures::CM_ASYNC_STACKFUL
+            | WasmFeatures::CM_THREADING
+            | WasmFeatures::CM_ERROR_CONTEXT;
+        if tunables.concurrency_support && !cfg!(feature = "component-model-async") {
+            bail!(
+                "concurrency support was requested but was not \
+                 compiled into this build of Wasmtime"
+            )
+        }
+        if !tunables.concurrency_support && features.intersects(requires_concurrency) {
+            bail!(
+                "concurrency support must be enabled to use the component \
+                 model async or threading features"
+            )
         }
 
         Ok((tunables, features))
@@ -2923,17 +2941,41 @@ impl Config {
         self
     }
 
-    #[cfg(feature = "component-model")]
-    #[inline]
-    pub(crate) fn cm_concurrency_enabled(&self) -> bool {
-        cfg!(feature = "component-model-async")
-            && self.enabled_features.intersects(
-                WasmFeatures::CM_ASYNC
-                    | WasmFeatures::CM_ASYNC_BUILTINS
-                    | WasmFeatures::CM_ASYNC_STACKFUL
-                    | WasmFeatures::CM_THREADING
-                    | WasmFeatures::CM_ERROR_CONTEXT,
-            )
+    /// Specifies whether support for concurrent execution of WebAssembly is
+    /// supported within this store.
+    ///
+    /// This configuration option affects whether runtime data structures are
+    /// initialized within a `Store` on creation to support concurrent execution
+    /// of WebAssembly guests. This is primarily applicable to the
+    /// [`Config::wasm_component_model_async`] configuration which is the first
+    /// time Wasmtime has supported concurrent execution of guests. This
+    /// configuration option, for example, enables usage of
+    /// [`Store::run_concurrent`], [`Func::call_concurrent`], [`StreamReader`],
+    /// etc.
+    ///
+    /// This configuration option can be manually disabled to avoid initializing
+    /// data structures in the [`Store`] related to concurrent execution. When
+    /// this option is disabled then APIs related to concurrency will all fail
+    /// with a panic. For example [`Store::run_concurrent`] will panic, creating
+    /// a [`StreamReader`] will panic, etc.
+    ///
+    /// The value of this option additionally affects whether a [`Config`] is
+    /// valid and the default set of enabled WebAssembly features. If this
+    /// option is disabled then component-model features related to concurrency
+    /// will all be disabled. If this option is enabled, then the options will
+    /// retain their normal defaults. It is not valid to create a [`Config`]
+    /// with component-model-async explicitly enabled and this option explicitly
+    /// disabled, however.
+    ///
+    /// This option defaults to `true`.
+    ///
+    /// [`Store`]: crate::Store
+    /// [`Store::run_concurrent`]: crate::Store::run_concurrent
+    /// [`Func::call_concurrent`]: crate::component::Func::call_concurrent
+    /// [`StreamReader`]: crate::component::StreamReader
+    pub fn concurrency_support(&mut self, enable: bool) -> &mut Self {
+        self.tunables.concurrency_support = Some(enable);
+        self
     }
 }
 

crates/wasmtime/src/engine/serialization.rs

@@ -285,7 +285,7 @@ impl Metadata<'_> {
             inlining_intra_module,
             inlining_small_callee_size,
             inlining_sum_size_threshold,
-            component_model_concurrency,
+            concurrency_support,
 
             // This doesn't affect compilation, it's just a runtime setting.
             memory_reservation_for_growth: _,
@@ -367,9 +367,9 @@ impl Metadata<'_> {
             "function inlining sum-size threshold",
         )?;
         Self::check_bool(
-            component_model_concurrency,
-            other.component_model_concurrency,
-            "component model concurrency",
+            concurrency_support,
+            other.concurrency_support,
+            "concurrency support",
         )?;
         Self::check_intra_module_inlining(inlining_intra_module, other.inlining_intra_module)?;
 

crates/wasmtime/src/runtime/component/concurrent.rs

@@ -886,8 +886,8 @@ impl<T> Store<T> {
         T: Send + 'static,
     {
         ensure!(
-            self.as_context().0.cm_concurrency_enabled(),
-            "cannot use `run_concurrent` without enabling component-model async"
+            self.as_context().0.concurrency_support(),
+            "cannot use `run_concurrent` when Config::concurrency_support disabled",
         );
         self.as_context_mut().run_concurrent(fun).await
     }
@@ -983,6 +983,11 @@ impl<T> StoreContextMut<'_, T> {
     /// This function can be used to invoke [`Func::call_concurrent`] for
     /// example within the async closure provided here.
     ///
+    /// This function will unconditionally return an error if
+    /// [`Config::concurrency_support`] is disabled.
+    ///
+    /// [`Config::concurrency_support`]: crate::Config::concurrency_support
+    ///
     /// # Store-blocking behavior
     ///
     /// At this time there are certain situations in which the `Future` returned
@@ -1056,8 +1061,8 @@ impl<T> StoreContextMut<'_, T> {
         T: Send + 'static,
     {
         ensure!(
-            self.0.cm_concurrency_enabled(),
-            "cannot use `run_concurrent` without enabling component-model async"
+            self.0.concurrency_support(),
+            "cannot use `run_concurrent` when Config::concurrency_support disabled",
         );
         self.do_run_concurrent(fun, false).await
     }
@@ -1080,7 +1085,7 @@ impl<T> StoreContextMut<'_, T> {
     where
         T: Send + 'static,
     {
-        debug_assert!(self.0.cm_concurrency_enabled());
+        debug_assert!(self.0.concurrency_support());
         check_recursive_run();
         let token = StoreToken::new(self.as_context_mut());
 
@@ -1513,8 +1518,10 @@ impl StoreOpaque {
     /// - The top-level instance is not already on the current task's call stack.
     /// - The instance is not in need of a post-return function call.
     /// - `self` has not been poisoned due to a trap.
-    pub(crate) fn may_enter_concurrent(&mut self, instance: RuntimeInstance) -> bool {
-        debug_assert!(self.cm_concurrency_enabled());
+    pub(crate) fn may_enter(&mut self, instance: RuntimeInstance) -> bool {
+        if !self.concurrency_support() {
+            return self.may_enter_at_all(instance);
+        }
         let state = self.concurrent_state_mut();
         if let Some(caller) = state.guest_thread {
             instance != state.get_mut(caller.task).unwrap().instance
@@ -1524,23 +1531,6 @@ impl StoreOpaque {
         }
     }
 
-    /// Returns `false` if the specified instance may not be entered, regardless
-    /// of what's on a task's call stack.
-    ///
-    /// If this returns `true`, the instance may be entered as long as it isn't
-    /// on the task's call stack, if applicable.
-    fn may_enter_at_all(&self, instance: RuntimeInstance) -> bool {
-        if self.trapped() {
-            return false;
-        }
-
-        let flags = self
-            .component_instance(instance.instance)
-            .instance_flags(instance.index);
-
-        unsafe { !flags.needs_post_return() }
-    }
-
     /// Variation of `may_enter` which takes a `TableId<GuestTask>` representing
     /// the callee.
     fn may_enter_task(&mut self, task: TableId<GuestTask>) -> bool {
@@ -1641,8 +1631,10 @@ impl StoreOpaque {
             .set_task_may_block(may_block)
     }
 
-    pub(crate) fn check_blocking_concurrent(&mut self) -> Result<()> {
-        debug_assert!(self.cm_concurrency_enabled());
+    pub(crate) fn check_blocking(&mut self) -> Result<()> {
+        if !self.concurrency_support() {
+            return Ok(());
+        }
         let state = self.concurrent_state_mut();
         let task = state.guest_thread.unwrap().task;
         let instance = state.get_mut(task).unwrap().instance.instance;

crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs

@@ -1116,14 +1116,20 @@ pub struct FutureReader<T> {
 
 impl<T> FutureReader<T> {
     /// Create a new future with the specified producer.
+    ///
+    /// # Panics
+    ///
+    /// Panics if [`Config::concurrency_support`] is not enabled.
+    ///
+    /// [`Config::concurrency_support`]: crate::Config::concurrency_support
     pub fn new<S: AsContextMut>(
         mut store: S,
         producer: impl FutureProducer<S::Data, Item = T>,
     ) -> Self
     where
         T: func::Lower + func::Lift + Send + Sync + 'static,
     {
-        assert!(store.as_context().0.cm_concurrency_enabled());
+        assert!(store.as_context().0.concurrency_support());
 
         struct Producer<P>(P);
 
@@ -1451,11 +1457,17 @@ where
     A: AsAccessor,
 {
     /// Create a new `GuardedFutureReader` with the specified `accessor` and `reader`.
+    ///
+    /// # Panics
+    ///
+    /// Panics if [`Config::concurrency_support`] is not enabled.
+    ///
+    /// [`Config::concurrency_support`]: crate::Config::concurrency_support
     pub fn new(accessor: A, reader: FutureReader<T>) -> Self {
         assert!(
             accessor
                 .as_accessor()
-                .with(|a| a.as_context().0.cm_concurrency_enabled())
+                .with(|a| a.as_context().0.concurrency_support())
         );
         Self {
             reader: Some(reader),
@@ -1503,14 +1515,20 @@ pub struct StreamReader<T> {
 
 impl<T> StreamReader<T> {
     /// Create a new stream with the specified producer.
+    ///
+    /// # Panics
+    ///
+    /// Panics if [`Config::concurrency_support`] is not enabled.
+    ///
+    /// [`Config::concurrency_support`]: crate::Config::concurrency_support
     pub fn new<S: AsContextMut>(
         mut store: S,
         producer: impl StreamProducer<S::Data, Item = T>,
     ) -> Self
     where
         T: func::Lower + func::Lift + Send + Sync + 'static,
     {
-        assert!(store.as_context().0.cm_concurrency_enabled());
+        assert!(store.as_context().0.concurrency_support());
         Self::new_(
             store
                 .as_context_mut()
@@ -1785,11 +1803,17 @@ where
 {
     /// Create a new `GuardedStreamReader` with the specified `accessor` and
     /// `reader`.
+    ///
+    /// # Panics
+    ///
+    /// Panics if [`Config::concurrency_support`] is not enabled.
+    ///
+    /// [`Config::concurrency_support`]: crate::Config::concurrency_support
     pub fn new(accessor: A, reader: StreamReader<T>) -> Self {
         assert!(
             accessor
                 .as_accessor()
-                .with(|a| a.as_context().0.cm_concurrency_enabled())
+                .with(|a| a.as_context().0.concurrency_support())
         );
         Self {
             reader: Some(reader),