wasmtime-wasi: Unexpected tcp stream break (Returning 0 for std::io::ErrorKind::WouldBlock)

[Heap-Hop]

wasmtime/crates/wasi/src/tcp.rs

Lines 719 to 723 in 6767488

	// Failing with `EWOULDBLOCK` is how we differentiate between a closed channel and no
	// data to read right now.
	Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => 0,

Returning 0 for std::io::ErrorKind::WouldBlock causes downstream to interpret it as a closed stream, see:
yoshuawuyts/wstd#25 (comment).
https://github.com/yoshuawuyts/wstd/blob/5ce367add5e7bcb569b6487453cb9ba94468dc99/src/io/copy.rs#L12

This is also found in:

wasmtime/crates/test-programs/src/bin/preview2_tcp_streams.rs

Lines 17 to 22 in 6767488

	// The input stream should immediately signal StreamError::Closed.
	// Notably, it should _not_ return an empty list (the wasi-io equivalent of EWOULDBLOCK)
	// See: https://github.com/bytecodealliance/wasmtime/pull/8968
	assert!(matches!(client.input.read(10), Err(StreamError::Closed)));

Given that the wit files already include many would-block errors, would it make sense to extend stream-error to include a would-block?

[pchickey]

WASI streams operations are nonblocking, unless they have blocking_ in their name, so read should always return an empty list instead of blocking. The blocking_ functions are equivalent to waiting for readiness on the stream's pollable, and then calling the nonblocking variant.

This is a significant departure from POSIX, which uses the file's mode to make the read and write syscalls blocking or nonblocking. So, while it makes sense for POSIX to use an error to indicate when a nonblocking socket would block, it doesn't make sense for WASI to follow that same convention.

[bjorn3]

How can you distinguish in WASI between an EOF stream (no data will be returned in the future because the connection was closed or the end of the file is reached) and a stream where you have to wait until the stream is ready before you can read more data? The distinction is very important for std::io::copy/tokio::io::copy which needs to continue reading until EOF and only returns when EOF is reached.

[pchickey]

When the stream is EOF, it returns Err(StreamError::Closed) as seen in the above snippet line 21. Maybe this isn't so much an "Error" as the "Err" constructor indicates, but the alternative was returning a custom enum there of Ok/Closed/Error, which we rejected on fairly arbitrary stylistic grounds.

[Heap-Hop]

Thanks for your kind explanation!

wasmtime/crates/wasi/wit/deps/io/streams.wit

Lines 47 to 54 in 6767488

	///
	/// This function returns a list of bytes containing the read data,
	/// when successful. The returned list will contain up to `len` bytes;
	/// it may return fewer than requested, but not more. The list is
	/// empty when no bytes are available for reading at this time. The
	/// pollable given by `subscribe` will be ready when more bytes are
	/// available.
	///

The pollable given by subscribe will be ready when more bytes are available.

I think it would be helpful to update the doc to explicitly note this situation for users.

[Heap-Hop]

WASI streams operations are nonblocking, unless they have blocking_ in their name

For the same reason, there's a potential bug in HostInputStream::blocking_read(also affects blocking_skip).
The doc states: it blocks until at least one byte can be read.

wasmtime/crates/wasi/src/stream.rs

Lines 23 to 28 in 6767488

	/// Similar to `read`, except that it blocks until at least one byte can be
	/// read.
	async fn blocking_read(&mut self, size: usize) -> StreamResult<Bytes> {
	self.ready().await;
	self.read(size)
	}

The ready() is implemented using readable from Tokio:

wasmtime/crates/wasi/src/tcp.rs

Lines 757 to 762 in 6767488

	#[async_trait::async_trait]
	impl Subscribe for TcpReadStream {
	async fn ready(&mut self) {
	self.0.lock().await.ready().await
	}
	}

wasmtime/crates/wasi/src/tcp.rs

Lines 739 to 745 in 6767488

	async fn ready(&mut self) {
	if self.closed {
	return;
	}
	self.stream.readable().await.unwrap();
	}

However, readable in Tokio does not await when io::ErrorKind::WouldBlock.

[Heap-Hop]

Maybe all we really need is to ensure that every ready() in WASI behaves as the explicit "ready" it's supposed to be.

[pchickey]

Thanks, it sounds like this is indeed a bug in wasmtime-wasi then. Can you create a minimal reproduction so I can work on a fix and use it as a test once we have the fix in? Something along the lines of https://github.com/bytecodealliance/wasmtime/blob/main/crates/test-programs/src/bin/preview2_tcp_streams.rs

[Heap-Hop]

@pchickey https://github.com/bytecodealliance/wasmtime/actions/runs/12070585313/job/33660435175?pr=9691#step:18:419

[pchickey]

Thanks! Its a holiday weekend here so I’ll get to this next week.

[alexcrichton]

I apologize it's taken me quite some time to take a look at this. Now that I've read over this and #9691 I personally don't think there's a bug in the wasmtime-wasi implementation here. I don't think WASI can reasonably provide super strict guarantees about exactly when things become ready and exactly the states before/after reads. Attempting to do this might not even be feasible in the face of an implementation across multiple platforms.

For example the test in #9691 looks like this:

        use test_programs::wasi::clocks::monotonic_clock::subscribe_duration;
        let timeout_100ms = 100_000_000;

        // Send some data to the server
        client.output.blocking_write_and_flush(b"Hi!").unwrap();

        server.input.subscribe().block();
        let res = server.input.read(512).unwrap();
        assert_eq!(res, b"Hi!", "Expected to receive data");

        // Don't send any data

        let res = server
            .input
            .subscribe()
            .block_until(&subscribe_duration(timeout_100ms));
        assert!(res.is_err(), "Expected to time out cause no data was sent");

I don't believe that this can be made to work reliably. Here after the initial "Hi!" message is ready it's not actually known whether the stream is readable or not. Determining this would involve actually consulting the kernel with some sort of syscall or something like that. It's a pretty important perf optimization that in async servers you don't do anything with the kernel until absolutely necessary.

For example here the kernel was consulted during the first server.input.subscribe().block(). At this point the stream is considered readable. The issued read then again consults the kernel and reads out "Hi!". At this point though the stream is notably considered still readable by Tokio and Wasmtime. It's not known whether this was a "short read" or whether more data is coming soon. This is why the next subscribe() immediately says "yes this is readable". Doing otherwise would require unnecessarily consulting the kernel.

Working robustly with async is certainly an art but personally I don't think that there's a bug here that needs fixing. I would personally think we can close #9691 and this issue as "working as intended".

[pchickey]

I apologize I never got back to this after saying I would. I started a new job and have a whole host of new problems to work on, so I dropped some of the old ones.

Thanks Alex for digging into this and helping me understand why the existing behavior is there. My conclusion is that my specification of "a ready pollable implies input-stream.read will succeed" is too strong of a guarantee to implement efficiently on top of POSIX. In discussion with Alex, he suggested that only the behavior of read shows whether input was ready or not, and the pollable's readiness is a best-effort indicator that provides an optimization over busy-looping on read. I expect that we can easily accommodate the weakening of that guarantee in existing code (at least in wstd), and any existing binaries can't possibly be relying on that guarantee since its never been upheld by implementations. We will have to take that discussion out to the spec repo in order to nail it all down.

[badeend]

I would personally think we can close (...) this issue as "working as intended".

For non-blocking APIs: agree. There's not much wasmtime can do (efficiently).

Maybe this could also justify the removal of starting an arbitrary-sized read in the FileInputStream implementation that was put in place to fulfill this guarantee?

---

For blocking APIs: there a few places where wasmtime blindly trusts that ready().await actually implies readiness. For example:

wasmtime/crates/wasi-io/src/streams.rs

Lines 24 to 29 in ca95576

	/// Similar to `read`, except that it blocks until at least one byte can be
	/// read.
	async fn blocking_read(&mut self, size: usize) -> StreamResult<Bytes> {
	self.ready().await;
	self.read(size)
	}

In these cases wasmtime is in a position to "fix" spurious wakeups by wrapping it in a loop.

[alexcrichton]

Yeah I would agree that blocking_read should have a loop to handle situations like this

[Heap-Hop]

This is why the next subscribe() immediately says "yes this is readable". Doing otherwise would require unnecessarily consulting the kernel.

@alexcrichton Thank you so much for the clear explanation, it resolved my confusion about why readable() in Tokio is not explicit.

It would be great if the blocking_ functions returned the expected results.

I would be happy to submit a PR to modify the blocking_ functions if no one else is currently working on it.

[alexcrichton]

That'd be most welcome, thank you!

[Heap-Hop]

#10113
Sorry, I haven't found an easy way to write a test that might block the process, as there isn’t a _until function for blocking_.
Is there an existing test case for this? The only approach I can think of is externally controlling the timeout for Wasmtime.