fix: add and use write_oneshot

Signed-off-by: Roman Volosatovs <rvolosatovs@riseup.net>

Author: rvolosatovs

crates/wasi/src/p3/sockets/host/types/tcp.rs

@@ -192,18 +192,6 @@ impl<T> AccessorTask<T, WasiSockets, wasmtime::Result<()>> for ListenTask {
     }
 }
 
-struct ResultWriteTask {
-    result: Result<(), ErrorCode>,
-    result_tx: FutureWriter<Result<(), ErrorCode>>,
-}
-
-impl<T> AccessorTask<T, WasiSockets, wasmtime::Result<()>> for ResultWriteTask {
-    async fn run(self, store: &Accessor<T, WasiSockets>) -> wasmtime::Result<()> {
-        self.result_tx.write(store, self.result).await;
-        Ok(())
-    }
-}
-
 struct ReceiveTask {
     stream: Arc<TcpStream>,
     data_tx: StreamWriter<Cursor<BytesMut>>,
@@ -250,17 +238,11 @@ impl<T> AccessorTask<T, WasiSockets, wasmtime::Result<()>> for ReceiveTask {
                 }
             }
         };
+        self.result_tx.write_oneshot(store, res);
         _ = self
             .stream
             .as_socketlike_view::<std::net::TcpStream>()
             .shutdown(Shutdown::Read);
-
-        // Write the result async from a separate task to ensure that all resources used by this
-        // task are freed
-        store.spawn(ResultWriteTask {
-            result: res,
-            result_tx: self.result_tx,
-        });
         Ok(())
     }
 }

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

@@ -535,6 +535,36 @@ impl<T> FutureWriter<T> {
         }
     }
 
+    /// Write the specified value to this `future`.
+    ///
+    /// Unlike [`write`](Self::write), this method ensures that the value has been written, but
+    /// does not wait for the receiver to receive it.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the store that the [`Accessor`] is derived from does not own
+    /// this future.
+    // TODO: Rework/refactor this as part of https://github.com/bytecodealliance/wasmtime/pull/11325
+    #[doc(hidden)]
+    pub fn write_oneshot(mut self, accessor: impl AsAccessor, value: T)
+    where
+        T: Send + 'static,
+    {
+        // FIXME: this is intended to be used in the future to directly
+        // manipulate state for this future within the store without having to
+        // go through an mpsc.
+        let _accessor = accessor.as_accessor();
+        let (tx, _) = oneshot::channel();
+        send(
+            &mut self.tx.as_mut().unwrap(),
+            WriteEvent::Write {
+                buffer: Some(value),
+                tx,
+            },
+        );
+        self.default = None;
+    }
+
     /// Wait for the read end of this `future` is dropped.
     ///
     /// The [`Accessor`] provided can be acquired from [`Instance::run_concurrent`] or