Skip to content
This repository was archived by the owner on Sep 8, 2025. It is now read-only.

Commit 4d344f6

Browse files
dicejdicej
committed
handle zero-length stream writes/reads per latest spec
Updated to match WebAssembly/component-model#444 Signed-off-by: Joel Dice <joel.dice@fermyon.com>
1 parent ab817cc commit 4d344f6

4 files changed

Lines changed: 150 additions & 34 deletions

File tree

‎crates/misc/component-async-tests/wit/test.wit‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -63,8 +63,10 @@ interface backpressure {
6363
interface transmit {
6464
variant control {
6565
read-stream(string),
66+
read-stream-zero,
6667
read-future(string),
6768
write-stream(string),
69+
write-stream-zero,
6870
write-future(string),
6971
}
7072

‎crates/test-programs/src/bin/async_transmit_callee.rs‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@ use {
1919
wit_future, wit_stream,
2020
},
2121
std::future::IntoFuture,
22-
wit_bindgen_rt::async_support::{self, FutureReader, StreamReader},
22+
wit_bindgen_rt::async_support::{self, FutureReader, StreamReader, StreamResult},
2323
};
2424

2525
struct Component;
@@ -48,6 +48,12 @@ impl Guest for Component {
4848
Control::ReadStream(value) => {
4949
assert_eq!(caller_stream_rx.next().await, Some(value));
5050
}
51+
Control::ReadStreamZero => {
52+
assert_eq!(
53+
caller_stream_rx.read(Vec::new()).await.0,
54+
StreamResult::Complete(0)
55+
);
56+
}
5157
Control::ReadFuture(value) => {
5258
assert_eq!(
5359
caller_future_rx1.take().unwrap().into_future().await,
@@ -57,6 +63,12 @@ impl Guest for Component {
5763
Control::WriteStream(value) => {
5864
assert!(callee_stream_tx.write_one(value).await.is_none());
5965
}
66+
Control::WriteStreamZero => {
67+
assert_eq!(
68+
callee_stream_tx.write(Vec::new()).await.0,
69+
StreamResult::Complete(0)
70+
);
71+
}
6072
Control::WriteFuture(value) => {
6173
callee_future_tx1
6274
.take()

‎crates/test-programs/src/bin/async_transmit_caller.rs‎

Lines changed: 61 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,13 +22,13 @@ use {
2222
local::local::transmit::{self, Control},
2323
wit_future, wit_stream,
2424
},
25-
futures::{future, FutureExt},
25+
futures::{future, stream::FuturesUnordered, FutureExt, StreamExt},
2626
std::{
2727
future::{Future, IntoFuture},
28-
pin::pin,
28+
pin::{pin, Pin},
2929
task::Poll,
3030
},
31-
wit_bindgen_rt::async_support::FutureWriteCancel,
31+
wit_bindgen_rt::async_support::{FutureWriteCancel, StreamResult},
3232
};
3333

3434
struct Component;
@@ -69,6 +69,38 @@ impl Guest for Component {
6969
.is_none());
7070
assert!(caller_stream_tx.write_one("c".into()).await.is_none());
7171

72+
// Tell the peer to do a zero-length read, do a zero-length write; assert the latter completes, then do a
73+
// non-zero-length write, assert that it does _not_ complete, then tell the peer to do a non-zero-length
74+
// read and assert that the write completes.
75+
assert!(control_tx
76+
.write_one(Control::ReadStreamZero)
77+
.await
78+
.is_none());
79+
{
80+
assert_eq!(
81+
caller_stream_tx.write(Vec::new()).await.0,
82+
StreamResult::Complete(0)
83+
);
84+
85+
let send = Box::pin(caller_stream_tx.write_one("d".into()));
86+
let Err(send) = poll(send).await else {
87+
panic!()
88+
};
89+
90+
let mut futures = FuturesUnordered::new();
91+
futures.push(Box::pin(send.map(|v| {
92+
assert!(v.is_none());
93+
})) as Pin<Box<dyn Future<Output = _>>>);
94+
futures.push(Box::pin(
95+
control_tx
96+
.write_one(Control::ReadStream("d".into()))
97+
.map(|v| {
98+
assert!(v.is_none());
99+
}),
100+
));
101+
while let Some(()) = futures.next().await {}
102+
}
103+
72104
// Start writing a value to the future, but cancel the write before telling the peer to read.
73105
{
74106
let send = Box::pin(caller_future_tx1.write("x".into()));
@@ -114,6 +146,32 @@ impl Guest for Component {
114146
.is_none());
115147
assert_eq!(callee_stream_rx.next().await, Some("b".into()));
116148

149+
// Tell the peer to do a zero-length write, assert that the read does _not_ complete, then tell the peer to
150+
// do a non-zero-length write and assert that the read completes.
151+
assert!(control_tx
152+
.write_one(Control::WriteStreamZero)
153+
.await
154+
.is_none());
155+
{
156+
let next = Box::pin(callee_stream_rx.next());
157+
let Err(next) = poll(next).await else {
158+
panic!()
159+
};
160+
161+
let mut futures = FuturesUnordered::new();
162+
futures.push(Box::pin(next.map(|v| {
163+
assert_eq!(v, Some("c".into()));
164+
})) as Pin<Box<dyn Future<Output = _>>>);
165+
futures.push(Box::pin(
166+
control_tx
167+
.write_one(Control::WriteStream("c".into()))
168+
.map(|v| {
169+
assert!(v.is_none());
170+
}),
171+
));
172+
while let Some(()) = futures.next().await {}
173+
}
174+
117175
// Start reading a value from the future, but cancel the read before telling the peer to write.
118176
{
119177
let next = Box::pin(callee_future_rx1.into_future());

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

Lines changed: 74 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -2231,6 +2231,21 @@ impl ComponentInstance {
22312231
ReadState::Open
22322232
};
22332233

2234+
let set_guest_ready = |me: &mut Self| {
2235+
let transmit = me.get_mut(transmit_id)?;
2236+
assert!(matches!(&transmit.write, WriteState::Open));
2237+
transmit.write = WriteState::GuestReady {
2238+
ty,
2239+
flat_abi,
2240+
options,
2241+
address: usize::try_from(address).unwrap(),
2242+
count: usize::try_from(count).unwrap(),
2243+
handle,
2244+
post_write: PostWrite::Continue,
2245+
};
2246+
Ok::<_, crate::Error>(())
2247+
};
2248+
22342249
let result = match mem::replace(&mut transmit.read, new_state) {
22352250
ReadState::GuestReady {
22362251
ty: read_ty,
@@ -2239,10 +2254,12 @@ impl ComponentInstance {
22392254
address: read_address,
22402255
count: read_count,
22412256
handle: read_handle,
2242-
..
22432257
} => {
22442258
assert_eq!(flat_abi, read_flat_abi);
22452259

2260+
let write_complete = count == 0 || read_count > 0;
2261+
let read_complete = count > 0;
2262+
22462263
let read_handle_rep = transmit.read_handle.rep();
22472264

22482265
let count = count.min(read_count);
@@ -2261,7 +2278,8 @@ impl ComponentInstance {
22612278
)?;
22622279

22632280
let code = ReturnCode::Completed(count.try_into().unwrap());
2264-
{
2281+
2282+
if read_complete {
22652283
self.push_event(
22662284
read_handle_rep,
22672285
match read_ty {
@@ -2277,9 +2295,24 @@ impl ComponentInstance {
22772295
},
22782296
},
22792297
)?;
2298+
} else {
2299+
let transmit = self.get_mut(transmit_id)?;
2300+
transmit.read = ReadState::GuestReady {
2301+
ty: read_ty,
2302+
flat_abi: read_flat_abi,
2303+
options: read_options,
2304+
address: read_address,
2305+
count: read_count,
2306+
handle: read_handle,
2307+
};
22802308
}
22812309

2282-
code
2310+
if write_complete {
2311+
code
2312+
} else {
2313+
set_guest_ready(self)?;
2314+
ReturnCode::Blocked
2315+
}
22832316
}
22842317

22852318
ReadState::HostReady { accept } => {
@@ -2295,19 +2328,7 @@ impl ComponentInstance {
22952328
}
22962329

22972330
ReadState::Open => {
2298-
assert!(matches!(&transmit.write, WriteState::Open));
2299-
2300-
let transmit = self.get_mut(transmit_id)?;
2301-
transmit.write = WriteState::GuestReady {
2302-
ty,
2303-
flat_abi,
2304-
options,
2305-
address: usize::try_from(address).unwrap(),
2306-
count: usize::try_from(count).unwrap(),
2307-
handle,
2308-
post_write: PostWrite::Continue,
2309-
};
2310-
2331+
set_guest_ready(self)?;
23112332
ReturnCode::Blocked
23122333
}
23132334

@@ -2370,6 +2391,20 @@ impl ComponentInstance {
23702391
WriteState::Open
23712392
};
23722393

2394+
let set_guest_ready = |me: &mut Self| {
2395+
let transmit = me.get_mut(transmit_id)?;
2396+
assert!(matches!(&transmit.read, ReadState::Open));
2397+
transmit.read = ReadState::GuestReady {
2398+
ty,
2399+
flat_abi,
2400+
options,
2401+
address: usize::try_from(address).unwrap(),
2402+
count: usize::try_from(count).unwrap(),
2403+
handle,
2404+
};
2405+
Ok::<_, crate::Error>(())
2406+
};
2407+
23732408
let result = match mem::replace(&mut transmit.write, new_state) {
23742409
WriteState::GuestReady {
23752410
ty: write_ty,
@@ -2384,6 +2419,9 @@ impl ComponentInstance {
23842419

23852420
let write_handle_rep = transmit.write_handle.rep();
23862421

2422+
let write_complete = write_count == 0 || count > 0;
2423+
let read_complete = write_count > 0;
2424+
23872425
let count = usize::try_from(count).unwrap().min(write_count);
23882426

23892427
self.copy(
@@ -2407,7 +2445,8 @@ impl ComponentInstance {
24072445
};
24082446

24092447
let code = ReturnCode::Completed(count.try_into().unwrap());
2410-
{
2448+
2449+
if write_complete {
24112450
self.push_event(
24122451
write_handle_rep,
24132452
match write_ty {
@@ -2421,9 +2460,25 @@ impl ComponentInstance {
24212460
},
24222461
},
24232462
)?;
2463+
} else {
2464+
let transmit = self.get_mut(transmit_id)?;
2465+
transmit.write = WriteState::GuestReady {
2466+
ty: write_ty,
2467+
flat_abi: write_flat_abi,
2468+
options: write_options,
2469+
address: write_address,
2470+
count: write_count,
2471+
handle: write_handle,
2472+
post_write,
2473+
};
24242474
}
24252475

2426-
code
2476+
if read_complete {
2477+
code
2478+
} else {
2479+
set_guest_ready(self)?;
2480+
ReturnCode::Blocked
2481+
}
24272482
}
24282483

24292484
WriteState::HostReady { accept, post_write } => {
@@ -2445,18 +2500,7 @@ impl ComponentInstance {
24452500
}
24462501

24472502
WriteState::Open => {
2448-
assert!(matches!(&transmit.read, ReadState::Open));
2449-
2450-
let transmit = self.get_mut(transmit_id)?;
2451-
transmit.read = ReadState::GuestReady {
2452-
ty,
2453-
flat_abi,
2454-
options,
2455-
address: usize::try_from(address).unwrap(),
2456-
count: usize::try_from(count).unwrap(),
2457-
handle,
2458-
};
2459-
2503+
set_guest_ready(self)?;
24602504
ReturnCode::Blocked
24612505
}
24622506

0 commit comments

Comments
 (0)