refactor {Stream,Future}|{Reader,Writer} APIs and internals

[dicej]

This makes a several changes to how {Stream,Future}|{Reader,Writer} work to make them more efficient and, in some ways, more ergonomic:

• The background tasks have been removed, allowing reads and writes to complete without task context switching. We now only allocate and use oneshot channels lazily when the other end is not yet ready; this improves real world performance benchmarks (e.g. wasi-http request handling) considerably.

• Instances of {Stream,Future}Reader can now be lifted and lowered directly; no need for Host{Stream,Future} anymore.

• The type parameter for Stream{Reader,Writer} no longer refers to the buffer type -- just the payload type (i.e. StreamReader<u8> instead of StreamReader<Vec<u8>>), meaning any buffer type may be used for a given read or write operation. This also means the compiler needs help with type inference less often when calling Instance::stream.

• Instances of {Stream,Future}|{Reader,Writer} now require access to the store in order to be disposed of properly. I've added RAII wrapper structs (WithAccessor[AndValue]) to help with this, and also updated Store::drop and Instance::run_concurrent to ensure the store thread-local is set when dropping futures closing over &Accessors.

• In order to ensure that resources containing {Stream,Future}|{Reader,Writer} instances are disposed of properly, I've added LinkerInstance::resource_concurrent and have updated wasmtime-wit-bindgen to use it. This gives resource drop functions access to a StoreContextMut via an Accessor, allowing the stream and future handles to be disposed of.

  • In order to make this work, I had to change Accessor::instance from a Instance to an Option<Instance>, which is awkward but temporary since we're planning to remove Accessor::instance entirely once we've moved concurrent state from ComponentInstance to Store.

That problem of disposal is definitely the most awkward part of all this. In simple cases, it's easy enough to ensure that read and write handles are disposed of properly, but both wasmtime-wasi and wasmtime-wasi-http have some pretty complicated functions where handles are passed between tasks and/or stored inside resources, so it can be tricky to ensure proper disposal on all code paths. I'm open to ideas for improving this, but I suspect we'll need new Rust language features (e.g. linear types) to make it truly ergonomic, robust, and efficient.

While testing the above, I discovered an issue with Instance::poll_until such that it would prematurely give up and return a "deadlock" trap error, believing that there was no further work to do, even though the future passed to it was ready to resolve the next time it was polled. I've fixed this by polling it one last time and only trapping if it returns pending.

Note that I've moved a few associated functions from ConcurrentState to Instance (e.g. guest_drop_writable and others) since they now need access to the store; they're unchanged otherwise. Apologies for the diff noise.

Finally, I've tweaked how wasmtime serve to poll the guest for content before handing the response to Hyper, which helps performance by ensuring the first content chunk can be sent with the same TCP packet as the beginning of the response.

[alexcrichton]

This is going to have conflicts with #11328, and I'll help work through those.

[dicej]

I've addressed all the feedback so far except that we're still lowering writes using same stack the host embedder called us on; I'll address that by using a fiber for lowering.

[alexcrichton]

Think about this in the context of rebasing with wasi:sockets -- is this necessary? Tasks can only be cancelled for two reasons:

1. The store is dropped - in this case there's no need to do further cleanup
2. The AbortHandle is triggered, but in this case it's statically known we just forget about that and ignore it.

Given that does WASI actually need Guarded*?

[dicej]

Perhaps we don't need it, but it saves us from having to remember to close the handle explicitly before returning, which isn't too onerous for this simple task, but could become a maintenance hazard later.

wasmtime-wasi-http has some pretty complicated control flow, where keeping track of what to close when without RAII would be a real pain.

[dicej]

@alexcrichton I believe I've addressed all your feedback so far. Let me know if you have anything else.

BTW, the optimization for lowering flat payloads during host writes is pretty trivial:

diff --git a/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs b/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs
index 2809cb6b12..ed56fd7bce 100644
--- a/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs
+++ b/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs
@@ -1777,9 +1777,7 @@ impl Instance {
                     anyhow::Ok(result)
                 };

-                if
-                // TODO: Check if payload is "flat"
-                false {
+                if T::IS_FLAT_TYPE {
                     // Optimize flat payloads (i.e. those which do not require
                     // calling the guest's realloc function) by lowering
                     // directly instead of using a oneshot::channel and
diff --git a/crates/wasmtime/src/runtime/component/func/typed.rs b/crates/wasmtime/src/runtime/component/func/typed.rs
index ad4328fce4..f27120957a 100644
--- a/crates/wasmtime/src/runtime/component/func/typed.rs
+++ b/crates/wasmtime/src/runtime/component/func/typed.rs
@@ -687,6 +687,9 @@ pub unsafe trait ComponentType: Send + Sync {
     #[doc(hidden)]
     const IS_RUST_UNIT_TYPE: bool = false;

+    #[doc(hidden)]
+    const IS_FLAT_TYPE: bool = false;
+
     /// Returns the number of core wasm abi values will be used to represent
     /// this type in its lowered form.
     ///
@@ -1016,6 +1019,8 @@ macro_rules! integers {

             const ABI: CanonicalAbiInfo = CanonicalAbiInfo::$abi;

+            const IS_FLAT_TYPE: bool = true;
+
             fn typecheck(ty: &InterfaceType, _types: &InstanceType<'_>) -> Result<()> {
                 match ty {
                     InterfaceType::$ty => Ok(()),