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

wasip3 TCP implementation #20

Merged
dicej merged 4 commits into from
Feb 21, 2025
Merged

wasip3 TCP implementation #20

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
dd73747
fix: only accept once guest is ready to receive
rvolosatovs Feb 21, 2025
1b89b8d
fix: ignore shutdown errors
rvolosatovs Feb 21, 2025
975fa29
feat: implement wasip3 TCP
rvolosatovs Feb 21, 2025
813996d
test: connect and accept concurrently
rvolosatovs Feb 21, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 15 additions & 10 deletions crates/test-programs/src/bin/sockets_0_3_tcp_bind.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,20 +61,25 @@ async fn test_tcp_bind_reuseaddr(ip: IpAddress) {

let connect_addr =
IpSocketAddress::new(IpAddress::new_loopback(ip.family()), bind_addr.port());
client.connect(connect_addr).await.unwrap();

let mut sock = accept.next().await.unwrap().unwrap();
assert_eq!(sock.len(), 1);
let sock = sock.pop().unwrap();
let (mut data_tx, data_rx) = wit_stream::new();
join!(
async {
sock.send(data_rx).await.unwrap();
client.connect(connect_addr).await.unwrap();
},
async {
data_tx.send(vec![0; 10]).await.unwrap();
drop(data_tx);
}
let mut sock = accept.next().await.unwrap().unwrap();
assert_eq!(sock.len(), 1);
let sock = sock.pop().unwrap();
let (mut data_tx, data_rx) = wit_stream::new();
join!(
async {
sock.send(data_rx).await.unwrap();
},
async {
data_tx.send(vec![0; 10]).await.unwrap();
drop(data_tx);
}
);
},
);

bind_addr
Expand Down
16 changes: 12 additions & 4 deletions crates/test-programs/src/bin/sockets_0_3_tcp_connect.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
use futures::{join, StreamExt as _};
use test_programs::p3::wasi::sockets::types::{
ErrorCode, IpAddress, IpAddressFamily, IpSocketAddress, TcpSocket,
};
Expand Down Expand Up @@ -96,12 +97,12 @@ async fn test_tcp_connect_dual_stack() {
async fn test_tcp_connect_explicit_bind(family: IpAddressFamily) {
let ip = IpAddress::new_loopback(family);

let listener = {
let (listener, mut accept) = {
let bind_address = IpSocketAddress::new(ip, 0);
let listener = TcpSocket::new(family);
listener.bind(bind_address).unwrap();
listener.listen().unwrap();
listener
let accept = listener.listen().unwrap();
(listener, accept)
};

let listener_address = listener.local_address().unwrap();
Expand All @@ -111,7 +112,14 @@ async fn test_tcp_connect_explicit_bind(family: IpAddressFamily) {
client.bind(IpSocketAddress::new(ip, 0)).unwrap();

// Connect should work:
client.connect(listener_address).await.unwrap();
join!(
async {
client.connect(listener_address).await.unwrap();
},
async {
accept.next().await.unwrap().unwrap();
}
);
}

impl test_programs::p3::exports::wasi::cli::run::Guest for Component {
Expand Down
111 changes: 56 additions & 55 deletions crates/test-programs/src/bin/sockets_0_3_tcp_sample_application.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,63 +20,64 @@ async fn test_tcp_sample_application(family: IpAddressFamily, bind_address: IpSo

let addr = listener.local_address().unwrap();

{
let client = TcpSocket::new(family);
client.connect(addr).await.unwrap();
let (mut data_tx, data_rx) = wit_stream::new();

join!(
async {
client.send(data_rx).await.unwrap();
},
async {
data_tx.send(vec![]).await.unwrap();
data_tx.send(first_message.into()).await.unwrap();
drop(data_tx);
}
);
}

{
let mut sock = accept.next().await.unwrap().unwrap();
assert_eq!(sock.len(), 1);
let sock = sock.pop().unwrap();

let (mut data_rx, fut) = sock.receive();
let data = data_rx.next().await.unwrap().unwrap();

// Check that we sent and received our message!
assert_eq!(data, first_message); // Not guaranteed to work but should work in practice.
fut.await.unwrap().unwrap().unwrap()
}
join!(
async {
let client = TcpSocket::new(family);
client.connect(addr).await.unwrap();
let (mut data_tx, data_rx) = wit_stream::new();
join!(
async {
client.send(data_rx).await.unwrap();
},
async {
data_tx.send(vec![]).await.unwrap();
data_tx.send(first_message.into()).await.unwrap();
drop(data_tx);
}
);
},
async {
let mut sock = accept.next().await.unwrap().unwrap();
assert_eq!(sock.len(), 1);
let sock = sock.pop().unwrap();

let (mut data_rx, fut) = sock.receive();
let data = data_rx.next().await.unwrap().unwrap();

// Check that we sent and received our message!
assert_eq!(data, first_message); // Not guaranteed to work but should work in practice.
fut.await.unwrap().unwrap().unwrap()
},
);

// Another client
{
let client = TcpSocket::new(family);
client.connect(addr).await.unwrap();
let (mut data_tx, data_rx) = wit_stream::new();
join!(
async {
client.send(data_rx).await.unwrap();
},
async {
data_tx.send(second_message.into()).await.unwrap();
drop(data_tx);
}
);
}

{
let mut sock = accept.next().await.unwrap().unwrap();
assert_eq!(sock.len(), 1);
let sock = sock.pop().unwrap();
let (mut data_rx, fut) = sock.receive();
let data = data_rx.next().await.unwrap().unwrap();

// Check that we sent and received our message!
assert_eq!(data, second_message); // Not guaranteed to work but should work in practice.
fut.await.unwrap().unwrap().unwrap()
}
join!(
async {
let client = TcpSocket::new(family);
client.connect(addr).await.unwrap();
let (mut data_tx, data_rx) = wit_stream::new();
join!(
async {
client.send(data_rx).await.unwrap();
},
async {
data_tx.send(second_message.into()).await.unwrap();
drop(data_tx);
}
);
},
async {
let mut sock = accept.next().await.unwrap().unwrap();
assert_eq!(sock.len(), 1);
let sock = sock.pop().unwrap();
let (mut data_rx, fut) = sock.receive();
let data = data_rx.next().await.unwrap().unwrap();

// Check that we sent and received our message!
assert_eq!(data, second_message); // Not guaranteed to work but should work in practice.
fut.await.unwrap().unwrap().unwrap()
}
);
}

impl test_programs::p3::exports::wasi::cli::run::Guest for Component {
Expand Down
30 changes: 21 additions & 9 deletions crates/test-programs/src/bin/sockets_0_3_tcp_sockopts.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
use futures::StreamExt as _;
use futures::{join, StreamExt as _};
use test_programs::p3::wasi::sockets::types::{
ErrorCode, IpAddress, IpAddressFamily, IpSocketAddress, TcpSocket,
};
Expand Down Expand Up @@ -132,10 +132,16 @@ async fn test_tcp_sockopt_inheritance(family: IpAddressFamily) {
let mut accept = listener.listen().unwrap();
let bound_addr = listener.local_address().unwrap();
let client = TcpSocket::new(family);
client.connect(bound_addr).await.unwrap();
let mut sock = accept.next().await.unwrap().unwrap();
assert_eq!(sock.len(), 1);
let sock = sock.pop().unwrap();
let ((), sock) = join!(
async {
client.connect(bound_addr).await.unwrap();
},
async {
let mut sock = accept.next().await.unwrap().unwrap();
assert_eq!(sock.len(), 1);
sock.pop().unwrap()
}
);

// Verify options on accepted socket:
{
Expand Down Expand Up @@ -194,10 +200,16 @@ async fn test_tcp_sockopt_after_listen(family: IpAddressFamily) {
}

let client = TcpSocket::new(family);
client.connect(bound_addr).await.unwrap();
let mut sock = accept.next().await.unwrap().unwrap();
assert_eq!(sock.len(), 1);
let sock = sock.pop().unwrap();
let ((), sock) = join!(
async {
client.connect(bound_addr).await.unwrap();
},
async {
let mut sock = accept.next().await.unwrap().unwrap();
assert_eq!(sock.len(), 1);
sock.pop().unwrap()
}
);

// Verify options on accepted socket:
{
Expand Down
12 changes: 10 additions & 2 deletions crates/test-programs/src/bin/sockets_0_3_tcp_states.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
use futures::{join, StreamExt as _};
use test_programs::p3::wasi::sockets::types::{
ErrorCode, IpAddress, IpAddressFamily, IpSocketAddress, TcpSocket,
};
Expand Down Expand Up @@ -146,10 +147,17 @@ async fn test_tcp_connected_state_invariants(family: IpAddressFamily) {
let bind_address = IpSocketAddress::new(IpAddress::new_loopback(family), 0);
let sock_listener = TcpSocket::new(family);
sock_listener.bind(bind_address).unwrap();
sock_listener.listen().unwrap();
let mut accept = sock_listener.listen().unwrap();
let addr_listener = sock_listener.local_address().unwrap();
let sock = TcpSocket::new(family);
sock.connect(addr_listener).await.unwrap();
join!(
async {
sock.connect(addr_listener).await.unwrap();
},
async {
accept.next().await.unwrap().unwrap();
}
);

assert_eq!(sock.bind(bind_address), Err(ErrorCode::InvalidState));
assert_eq!(
Expand Down
Loading